Apache Airflow Concepts Avancés : Guide des Opérations par Version avec Exemples Pratiques

Introduction

Apache Airflow est une plateforme puissante d'orchestration de workflows, mais ses concepts avancés restent souvent opaques pour les opérateurs qui savent uniquement écrire des DAG. Ce guide explique les concepts avancés à travers des exemples pratiques : inventaire des versions et de l'environnement, chemins de configuration sûrs, vérification et diagnostics, modes de défaillance et récupération. Nous mettons l'accent sur la sécurité opérationnelle : observer avant de modifier, limiter le rayon d'impact, utiliser des variables à la place des secrets, vérifier le résultat et documenter la procédure de récupération si l'état attendu n'est pas atteint.

Cet article s'adresse aux développeurs, consultants DevOps et équipes techniques de startups qui doivent passer d'un problème observé à un résultat vérifié. Il relie les rouages internes d'Apache Airflow, son architecture, les sujets approfondis et des exemples à des commandes, des sorties attendues, des signaux de défaillance et des décisions de récupération. Nous couvrons à la fois Airflow 2 et Airflow 3, en soulignant les différences clés comme le processeur de DAG séparé et le serveur API dans Airflow 3.

À la fin, vous disposerez de procédures concrètes pour inspecter, configurer et récupérer des déploiements Airflow, chaque recommandation étant liée à une portée de version et à un signal de vérification.

Inventaire des versions et de l'environnement

Avant toute intervention, vous devez savoir exactement ce que vous exécutez. Cette section fournit un processus d'inventaire en lecture seule pour capturer le composant concerné, la plage de versions, les prérequis et l'état actuel sans apporter de modifications.

Identifier votre version d'Airflow et la topologie

Exécutez la commande suivante pour obtenir la version d'Airflow :

airflow version

Exemple de sortie attendue :

2.10.3

Pour Airflow 3, la sortie peut être 3.0.0 ou une version ultérieure. Le numéro de version détermine les commandes et les points de terminaison de contrôle de santé disponibles. Documentez toujours la version avant de procéder au dépannage.

Ensuite, déterminez l'exécuteur et la base de données de métadonnées. Vous pouvez inspecter la configuration sans la modifier :

airflow config get-value core executor
airflow config get-value database sql_alchemy_conn

Exemple de sortie :

CeleryExecutor
postgresql+psycopg2://airflow:airflow@postgres/airflow

Remarque : En production, évitez d'afficher la chaîne de connexion SQLAlchemy complète si elle contient un mot de passe. Utilisez airflow config get-value database sql_alchemy_conn uniquement dans un environnement sécurisé, ou masquez la section du mot de passe lors de la journalisation. Mieux encore, utilisez des variables d'environnement et ne codez jamais de secrets en dur dans airflow.cfg.

Pour la méthode de déploiement, vérifiez si vous utilisez Docker, Kubernetes ou une installation autonome :

# Docker
docker ps --filter "name=airflow"

# Kubernetes
kubectl get pods -n airflow

Si vous êtes sur Kubernetes, vérifiez également la version du chart Helm :

helm list -n airflow

Enregistrez ces sorties avec un horodatage. Par exemple :

date -u +"%Y-%m-%dT%H:%M:%SZ" && airflow version

Airflow 2 vs Airflow 3 : différences de composants

Airflow 3 introduit un composant processeur de DAG séparé et un serveur API. Dans Airflow 2, le planificateur analyse souvent les DAG dans le processus. Dans Airflow 3, le processeur de DAG s'exécute comme son propre processus, et le serveur API gère les requêtes REST. Cela affecte les contrôles de santé et l'allocation des ressources.

Pour voir les processus en cours d'exécution, utilisez :

ps aux | grep -E "airflow (scheduler|dag-processor|api-server|webserver)"

Attendez-vous à des lignes de sortie pour chaque composant. Dans Airflow 3, vous devriez voir des processus séparés pour le processeur de DAG et le serveur API. Si vous ne voyez qu'un planificateur et un serveur Web, vous êtes probablement sur Airflow 2.

Contrôles de santé en lecture seule avant d'intervenir

Les contrôles de santé doivent être en lecture seule et adaptés à la version. Dans Airflow 2, la commande airflow jobs check fonctionne pour les contrôles du planificateur. Dans Airflow 3, utilisez le point de terminaison de santé du serveur API.

Pour Airflow 2, vérifiez la santé du planificateur :

airflow jobs check --job-type SchedulerJob --hostname $(hostname)

Sortie attendue en cas de succès :

Found one alive job.

Pour Airflow 3, interrogez le point de terminaison de santé du serveur API :

curl -s http://localhost:8080/api/v2/monitor/health

Exemple de réponse :

{
  "metadatabase": {
    "status": "healthy"
  },
  "scheduler": {
    "status": "healthy",
    "latest_scheduler_heartbeat": "2024-01-01T00:00:00Z"
  },
  "triggerer": {
    "status": "healthy",
    "latest_triggerer_heartbeat": "2024-01-01T00:00:00Z"
  },
  "dag_processor": {
    "status": "healthy",
    "latest_heartbeat": "2024-01-01T00:00:00Z"
  }
}

Ne considérez pas le code HTTP 200 comme une preuve que chaque composant est sain. Inspectez chaque champ. Par exemple, si scheduler.status est "unhealthy" mais que le statut HTTP est 200, le système global peut encore être dégradé.

Vérifiez la connectivité de la base de données de métadonnées avec :

airflow db check

Sortie attendue :

Connection successful.

Si la commande échoue, vous verrez une erreur comme sqlalchemy.exc.OperationalError. Corrigez la connexion à la base de données avant de continuer.

Considérations de mise à niveau (Airflow 2.7+ et 3.x)

Lors d'une mise à niveau, ne présumez jamais qu'un redémarrage suffit. Suivez ces étapes :

1. Consultez les notes de version de la version cible.
2. Vérifiez la compatibilité des paquets fournisseurs. Utilisez pip list | grep apache-airflow-providers pour lister les fournisseurs installés.
3. Sauvegardez la base de données de métadonnées. Pour PostgreSQL :

pg_dump -U airflow airflow > airflow_backup_$(date +%Y%m%d).sql

1. Testez des DAG représentatifs dans un environnement de préproduction avant la mise à niveau en production.
2. Exécutez la migration de la base de données :

airflow db migrate

N'exécutez pas airflow db upgrade ni ne redémarrez les services avant la fin de la migration. La commande db migrate applique les changements de schéma et peut prendre du temps selon la taille de la base de données.

Chemin de configuration sûr

Les changements de configuration sont une source fréquente de pannes. Cette section explique comment observer la configuration actuelle, apporter des modifications minimales et vérifier le résultat.

Inspection de la configuration en lecture seule

Avant de modifier une configuration, inspectez les valeurs effectives actuelles. Utilisez :

airflow config list

Cela affiche toutes les options de configuration avec leurs valeurs actuelles, mais attention : cela peut inclure des secrets. À la place, récupérez des paramètres spécifiques :

airflow config get-value core dags_folder
airflow config get-value scheduler scheduler_heartbeat_sec

Exemple de sortie :

/opt/airflow/dags
5

Pour les variables d'environnement qui remplacent la configuration, vérifiez avec :

env | grep AIRFLOW__

La sortie attendue comprend des lignes comme :

AIRFLOW__CORE__EXECUTOR=CeleryExecutor
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow

Encore une fois, masquez les secrets dans les journaux. Utilisez un outil comme sed pour masquer les mots de passe :

env | grep AIRFLOW__ | sed 's/://[^:]*:[^@]*@/://***:***@/'

Exemple de modification minimale : augmenter le battement de cœur du planificateur

Supposons que vous observez que le planificateur manque des battements de cœur, ce qui retarde l'exécution des DAG. Dans Airflow 2, la valeur par défaut de scheduler_heartbeat_sec est de 5 secondes. Vous pourriez vouloir l'augmenter pour réduire la charge sur la base de données. Cependant, l'augmenter trop peut retarder la détection d'une défaillance du planificateur.

Pour apporter une modification minimale, sauvegardez d'abord le fichier de configuration actuel :

cp /opt/airflow/airflow.cfg /opt/airflow/airflow.cfg.bak.$(date +%Y%m%d)

Ensuite, modifiez airflow.cfg sous [scheduler] :

[scheduler]
scheduler_heartbeat_sec = 10

Alternativement, définissez une variable d'environnement avant de démarrer le planificateur :

export AIRFLOW__SCHEDULER__SCHEDULER_HEARTBEAT_SEC=10

Après avoir redémarré le planificateur, vérifiez que la modification a pris effet :

airflow config get-value scheduler scheduler_heartbeat_sec

Sortie attendue : 10.

Ensuite, vérifiez à nouveau la santé du planificateur avec airflow jobs check --job-type SchedulerJob --hostname $(hostname) ou l'API de santé dans Airflow 3.

Gestion des secrets de configuration

Ne stockez jamais de secrets directement dans airflow.cfg. Utilisez des variables d'environnement ou un backend de secrets. Par exemple, définissez le mot de passe de la base de données via une variable d'environnement :

export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:${DB_PASSWORD}@postgres/airflow

En production, utilisez un gestionnaire de secrets comme HashiCorp Vault ou AWS Secrets Manager. Airflow prend en charge les backends de secrets via la configuration airflow.secrets.

Pour tester si les secrets sont résolubles, exécutez un DAG simple qui affiche une URI de connexion sans révéler le mot de passe. Mieux encore, utilisez airflow connections get avec une sortie masquée. Par exemple :

airflow connections get my_postgres_conn

Si le mot de passe est affiché, assurez-vous que votre backend de secrets le masque.

Vérification et diagnostics

La vérification consiste à confirmer que le système se comporte comme prévu. Cette section fournit des commandes et des sorties attendues pour les scénarios de diagnostic courants.

Vérification de l'analyse et de la planification des DAG

Airflow analyse les fichiers DAG pour détecter les erreurs et mettre à jour le sac de DAG. Dans Airflow 2, le planificateur analyse les DAG ; dans Airflow 3, le processeur de DAG le fait. Pour vérifier les erreurs d'analyse sans déclencher d'exécutions :

airflow dags list

La sortie attendue est un tableau des identifiants de DAG, des propriétaires et des intervalles de planification. Si un DAG comporte des erreurs, il peut ne pas apparaître ou afficher un nombre d'erreurs.

Pour voir les détails d'un DAG spécifique :

airflow dags show example_dag

Cela affiche la structure du DAG. Si le DAG ne peut pas être analysé, vous verrez une erreur.

Dans Airflow 3, vous pouvez interroger le serveur API pour voir l'état du traitement des DAG :

curl -s http://localhost:8080/api/v2/dags?limit=100 | jq '.dags[] | {dag_id, is_paused, fileloc}'

Cela renvoie du JSON avec l'identifiant de chaque DAG, l'état de pause et l'emplacement du fichier.

Analyse approfondie de la santé du planificateur

Le planificateur est critique. Dans Airflow 2, utilisez airflow jobs check comme indiqué précédemment. De plus, inspectez les journaux du planificateur :

tail -n 100 /opt/airflow/logs/scheduler/latest/scheduler.log

Recherchez des erreurs comme Scheduler heartbeat timed out ou Unable to acquire lock on database.

Dans Airflow 3, le planificateur et le processeur de DAG ont des journaux séparés. Vérifiez les journaux du processeur de DAG :

tail -n 100 /opt/airflow/logs/dag_processor/latest/dag_processor.log

Connectivité et performances de la base de données

Utilisez airflow db check pour vérifier la connectivité. Pour les problèmes de performances, interrogez directement la base de données. Par exemple, pour voir le nombre de tâches en file d'attente :

psql -U airflow -d airflow -c "SELECT state, count(*) FROM task_instance GROUP BY state;"

Exemple de sortie attendue :

   state   | count
-----------+-------
 queued    |    15
 running   |     3
 success   |  1200
 failed    |     2

Si le nombre en file d'attente augmente avec le temps, le planificateur peut ne pas traiter les tâches assez rapidement. Vérifiez la capacité des travailleurs ou la configuration du planificateur.

Santé du serveur API dans Airflow 3

Le serveur API d'Airflow 3 est un composant séparé. Vérifiez sa santé :

curl -s http://localhost:8080/api/v2/monitor/health

Comme indiqué précédemment, la réponse inclut les états des composants. Si le serveur API est en panne, le contrôle de santé échouera avec une erreur de connexion. Assurez-vous que le processus du serveur API est en cours d'exécution :

ps aux | grep "airflow api-server"

Modes de défaillance et récupération

Comprendre les modes de défaillance courants et comment récupérer est essentiel pour garder Airflow fiable. Cette section couvre des défaillances spécifiques, leurs symptômes et les étapes de récupération.

Planificateur en panne ou battement de cœur perdu

Symptômes : les exécutions de DAG ne sont pas déclenchées, les tâches restent dans l'état queued, et airflow jobs check ne signale aucun job actif.

Étapes de récupération :

1. Vérifiez si le processus du planificateur est en cours d'exécution.
2. Sinon, démarrez-le : airflow scheduler (dans Airflow 2) ou airflow scheduler et airflow dag-processor (dans Airflow 3).
3. S'il est en cours d'exécution mais ne bat pas, inspectez les journaux pour des erreurs de connexion à la base de données. Corrigez la connexion à la base de données.
4. Vérifiez avec airflow jobs check ou l'API de santé.

Panne de la base de données de métadonnées

Symptômes : airflow db check échoue, le planificateur et le serveur Web journalisent des erreurs de connexion à la base de données.

Étapes de récupération :

1. Vérifiez l'état du serveur de base de données : pg_isready -h postgres -p 5432 pour PostgreSQL.
2. S'il est en panne, redémarrez le service de base de données.
3. Vérifiez qu'Airflow peut se connecter : airflow db check.
4. Si la base de données est en ligne mais qu'Airflow ne peut toujours pas se connecter, vérifiez le réseau, les informations d'identification et le pare-feu.
5. Une fois restaurée, assurez-vous que le planificateur reprend son battement de cœur.

Processeur de DAG bloqué (Airflow 3)

Symptômes : les nouveaux DAG n'apparaissent pas, les modifications des fichiers DAG ne sont pas reflétées, et le point de terminaison de santé affiche dag_processor.status: unhealthy.

Étapes de récupération :

1. Vérifiez les journaux du processeur de DAG pour des erreurs.
2. Redémarrez le processeur de DAG : airflow dag-processor.
3. Vérifiez la santé : curl -s http://localhost:8080/api/v2/monitor/health | jq '.dag_processor.status' devrait renvoyer "healthy".

Échecs de tâches et reprises

Symptômes : les tâches échouent avec des erreurs spécifiques. La récupération dépend de l'erreur. Vérifiez toujours d'abord les journaux de la tâche :

airflow tasks logs example_dag example_task 2024-01-01T00:00:00+00:00

Si l'erreur est transitoire (par exemple, un délai d'attente réseau), relancez la tâche :

airflow tasks retry example_dag example_task 2024-01-01T00:00:00+00:00

Pour une correction permanente, mettez à jour le code du DAG et réexécutez.

Récupération après un échec de mise à niveau

Si une mise à niveau échoue, ne tentez pas de redémarrer avec la nouvelle version. Revenez à la version précédente et restaurez la sauvegarde de la base de données.

Étapes :

1. Arrêtez les services Airflow.
2. Restaurez la base de données à partir de la sauvegarde : psql -U airflow -d airflow < airflow_backup_20240101.sql
3. Réinstallez la version précédente d'Airflow : pip install apache-airflow==2.10.3
4. Démarrez les services et vérifiez avec les contrôles de santé.

Ayez toujours un plan de retour en arrière testé avant une mise à niveau.

Liste de contrôle des opérations

Utilisez cette liste de contrôle pour toute opération Airflow, qu'il s'agisse d'un changement de configuration, d'une mise à niveau ou d'une session de dépannage.

Avant tout changement

• [ ] Enregistrez la version actuelle d'Airflow : exécutez airflow version et notez la sortie.
• [ ] Identifiez la topologie de déploiement : Docker, Kubernetes ou autonome. Vérifiez avec docker ps ou kubectl get pods -n airflow.
• [ ] Notez l'exécuteur et la base de données de métadonnées : airflow config get-value core executor et airflow config get-value database sql_alchemy_conn (masquez le mot de passe).
• [ ] Capturez l'état de santé actuel : exécutez airflow jobs check (Airflow 2) ou curl http://localhost:8080/api/v2/monitor/health (Airflow 3). Enregistrez la sortie.
• [ ] Sauvegardez la base de données de métadonnées : pg_dump -U airflow airflow > airflow_backup_$(date +%Y%m%d).sql
• [ ] Examinez les journaux récents pour des erreurs : tail -n 200 /opt/airflow/logs/scheduler/latest/scheduler.log
• [ ] Assurez-vous d'avoir un plan de retour en arrière.

Pendant le changement

• [ ] Ne modifiez qu'un seul paramètre à la fois.
• [ ] Utilisez des variables d'environnement ou un backend de secrets pour les valeurs sensibles ; ne codez jamais de secrets en dur.
• [ ] Si vous modifiez airflow.cfg, créez d'abord une sauvegarde : cp /opt/airflow/airflow.cfg /opt/airflow/airflow.cfg.bak.$(date +%Y%m%d)
• [ ] Appliquez le changement à un environnement de test d'abord, si possible.

Après le changement

• [ ] Vérifiez que le changement a pris effet : airflow config get-value <section> <key> devrait renvoyer la nouvelle valeur.
• [ ] Vérifiez à nouveau la santé : airflow jobs check ou le point de terminaison de l'API de santé.
• [ ] Surveillez les journaux pour de nouvelles erreurs pendant au moins quelques minutes.
• [ ] Confirmez que l'analyse des DAG est réussie : airflow dags list ne devrait pas afficher d'erreurs.
• [ ] Si le changement implique une migration de base de données, exécutez airflow db migrate avant de démarrer les services.
• [ ] Documentez le changement et les résultats de vérification, y compris les horodatages.

Vérification de la récupération

• [ ] Si un retour en arrière a été effectué, vérifiez que le système est revenu à l'état stable précédent en vérifiant la version, la configuration et la santé.
• [ ] Confirmez qu'aucune donnée n'a été perdue : vérifiez les instances de tâches récentes dans la base de données.

Conclusion

Les concepts avancés d'Apache Airflow deviennent gérables lorsque vous appliquez des procédures versionnées, observables et réversibles. Chaque recommandation de ce guide inclut une commande ou un signal pour vérifier le résultat. Commencez par identifier votre version d'Airflow et la topologie, puis utilisez des contrôles de santé en lecture seule avant d'apporter tout changement. Limitez les changements à un élément défini, protégez les secrets et ayez toujours un plan de récupération.

Comme prochaine étape, choisissez une vérification à faible risque de cet article, comme vérifier le point de terminaison de santé dans Airflow 3 ou exécuter airflow db check dans Airflow 2. Enregistrez l'état actuel, exécutez la vérification documentée et comparez le résultat avec le signal attendu. Examinez les dépendances telles que les paquets fournisseurs et la connectivité de la base de données.

Un flux de travail technique fiable rend les échecs visibles, protège les valeurs sensibles, limite les changements à la ressource prévue et définit la vérification de la récupération avant qu'un incident ne force la décision. Avec ces pratiques, vous pouvez garder votre déploiement Airflow sain et résilient.