Nettoyer la base de données et les logs de Apache Airflow
Introduction
Apache Airflow est un outil puissant pour l'orchestration de workflows, mais comme tout système de gestion de tâches, il a besoin d’un peu d’entretien pour rester performant. Si vous utilisez Airflow en production ou sur de longues périodes, vous avez probablement remarqué une croissance importante de sa base de données et de ses fichiers de logs. Voici comment effectuer un bon ménage !
Faire du ménage
Airflow stocke une quantité importante de métadonnées dans sa base de données : exécutions de DAGs, tâches, logs, événements, etc. Avec le temps, cela peut :
- Ralentir l’interface web (UI)
- Allonger les temps de chargement de DAGs
- Consommer beaucoup d’espace disque
De plus, les fichiers de log générés pour chaque tâche peuvent s’accumuler rapidement sur le disque.
Clean Database
Airflow fournit une commande pour supprimer les anciennes métadonnées : airflow db clean. Cette commande permet de supprimer les exécutions de DAGs, les journaux d’événements, les tâches obsolètes, etc.
##########################################################################
# PARAMÈTRES
NOMBRE_MOIS_A_CONSERVER = 1 # <- À adapter
NOMBRE_JOUR_A_CONSERVER = NOMBRE_MOIS_A_CONSERVER*30 # <- À adapter
print(f"Nombre de mois à conserver : {NOMBRE_MOIS_A_CONSERVER}")
print(f"Nombre de jours à conserver : {NOMBRE_JOUR_A_CONSERVER}")
##########################################################################
# CALCUL DE LA DATE LIMITE
now = pendulum.now()
clean_before = now.subtract(months=NOMBRE_MOIS_A_CONSERVER)
clean_before_str = clean_before.format("YYYY-MM-DDTHH:mm:ss")
clean_db = BashOperator(
task_id="clean_db",
bash_command=f"airflow db clean --clean-before-timestamp {clean_before_str} --yes --skip-archive",
)Cela supprimera toutes les métadonnées vieilles de plus de 1 mois.
- Cette opération est irréversible. Sauvegardez votre base si nécessaire.
- Adapter selon votre besoin.
Quelques informations supplémentaires sur la commande airflow db clean.
- https://airflow.apache.org/docs/apache-airflow/stable/howto/set-up-database.html#database-cleaning-tools
- https://fig.io/manual/airflow/db/clean
Suppression logs
Par défaut, les logs sont stockés dans $AIRFLOW_HOME/logs/. Ils peuvent être supprimés manuellement ou automatiquement avec un script.
Voici comment supprimer les logs et afficher le résultat :
clean_logs = BashOperator(
task_id="delete_log_files",
bash_command=(
"for f in $(find $AIRFLOW_HOME/logs -type f -mtime +{{ params.jours }}); do "
"echo \"Suppression de : $f\"; rm -f \"$f\"; "
"done"
),
params={"jours": NOMBRE_JOUR_A_CONSERVER},
)Pour supprimer les dossiers vides après nettoyage des fichiers de logs :
clean_folders = BashOperator(
task_id="delete_empty_folders",
bash_command=(
"echo 'Dossiers vides supprimés :' && "
"find $AIRFLOW_HOME/logs -type d -empty -print -delete"
),
)Selon votre configuration Airflow, votre nombre de DAGs actifs, les tasks de suppression des logs et des dossiers vides peuvent durer plusieurs minutes.
Dag de maintenance
Ci-dessous mon dag de maintenance cleaning.py.
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from datetime import datetime, timedelta
import pendulum
##########################################################################
# PARAMÈTRES
NOMBRE_MOIS_A_CONSERVER = 1
NOMBRE_JOUR_A_CONSERVER = NOMBRE_MOIS_A_CONSERVER*30
print(f"Nombre de mois à conserver : {NOMBRE_MOIS_A_CONSERVER}")
print(f"Nombre de jours à conserver : {NOMBRE_JOUR_A_CONSERVER}")
##########################################################################
# CALCUL DE LA DATE LIMITE
now = pendulum.now()
clean_before = now.subtract(months=NOMBRE_MOIS_A_CONSERVER)
clean_before_str = clean_before.format("YYYY-MM-DDTHH:mm:ss")
DAG_ID = 'cleaning'
DAG_NAME = 'Maintenance Db et Log'
DAG_TAG = ["maintenance","db","log"]
TIMEZONE = env.TIMEZONE
default_args_dag = {
'owner' : 'ndl',
'description' : DAG_NAME,
'depend_on_past' : False,
'start_date' : datetime(2025, 5, 5), #pendulum.today(TIMEZONE).add(days=-1), # annee / mois / jour
'email_on_failure' : False,
'email_on_retry' : False,
'retries' : 0,
'retry_delay' : timedelta(minutes=1),
'dagrun_timeout' : timedelta(minutes=300),
}
with DAG(DAG_ID,
dag_display_name=DAG_NAME,
default_args=default_args_dag,
catchup=False,
tags=DAG_TAG,
schedule_interval=None,#"3 10 * * 1-5", # https://crontab.cronhub.io/
is_paused_upon_creation=True,
on_failure_callback=failure_callbacks,
on_success_callback=None,
max_active_runs=1,
) as dag:
clean_db = BashOperator(
task_id="clean_db",
bash_command=f"airflow db clean --clean-before-timestamp {clean_before_str} --yes",
)
clean_logs = BashOperator(
task_id="delete_log_files",
bash_command=(
"for f in $(find $AIRFLOW_HOME/logs -type f -mtime +{{ params.jours }}); do "
"echo \"Suppression de : $f\"; rm -f \"$f\"; "
"done"
),
params={"jours": NOMBRE_JOUR_A_CONSERVER},
)
clean_folders = BashOperator(
task_id="delete_empty_folders",
bash_command=(
"echo 'Dossiers vides supprimés :' && "
"find $AIRFLOW_HOME/logs -type d -empty -print -delete"
),
)
clean_db >> clean_logs >> clean_foldersRésumé
Un peu d’entretien permet de garder votre environnement Apache Airflow sain et rapide. Entre la purge des anciennes exécutions et le nettoyage des logs, ces actions simples prolongent la vie de votre orchestrateur sans douleur. Pensez à les intégrer à votre routine !
En supplément, Il faudrait réaliser une ré-indexation de la base airflow et un nettoyage avec "VACUUM" ou "VACUUM ANALYSE" ou "ANALYSE" pour postgre sql afin de mettre à jour les statistiques en sus. Le vacuum se fait à intervalle régulier par défaut normalement.
REINDEX DATABASE airflow_db;
VACUUM ANALYZE;
VACUUM;
ANALYSE;
