En production, un pipeline de données ne s'exécute pas toujours de manière linéaire et prévisible. Une API externe peut tomber temporairement en panne, un schéma de données source peut évoluer de manière inattendue, ou la direction peut vous demander de recalculer les métriques des six derniers mois suite à un changement de règle comptable.
Pour faire face à ces situations réelles, Apache Airflow fournit des outils avancés d'intervention manuelle, de débogage unitaire et de rattrapage historique (Backfill).
1. Déclencher et piloter depuis l'interface Web (UI)#
L'interface graphique offre un contrôle direct sur l'état de santé de chaque exécution.
flowchart LR
A["Bouton Play (Déclenchement)"] --> B["<b>Trigger DAG</b><br>Lance l'exécution avec la date actuelle"]
A --> C["<b>Trigger DAG w/ config</b><br>Passe des paramètres JSON personnalisés"]
D["Clic sur une Tâche Échouée"] --> E["<b>Clear (Effacer l'état)</b><br>Remet la tâche en file d'attente"]
D --> F["<b>Mark Success / Failed</b><br>Force artificiellement un statut"]
style A fill:#f1f5f9,stroke:#475569
style B fill:#e0f2fe,stroke:#0284c7
style C fill:#fed7aa,stroke:#ea580c
style D fill:#fee2e2,stroke:#ef4444
style E fill:#dcfce7,stroke:#16a34a,stroke-width:2px,color:#0f172a
style F fill:#f1f5f9,stroke:#64748b
Déclencher avec une configuration personnalisée (Trigger w/ config)#
Lorsque vous cliquez sur Trigger DAG w/ config, vous pouvez injecter un dictionnaire JSON qui sera lu dynamiquement par vos tâches :
{
"date_cible": "2025-12-31",
"mode_force": true,
"pays_filtre": ["France", "Allemagne", "Italie"]
}
Dans votre code Python, ces variables sont immédiatement récupérables via l'objet de contexte dag_run.conf :
def traiter_payload(**context):
conf = context["dag_run"].conf or {}
mode_force = conf.get("mode_force", False)
date_cible = conf.get("date_cible", context["ds"])
print(f"Exécution pour {date_cible} avec mode_force={mode_force}")
Le mécanisme magique du "Clear" (Relancer sans tout réexécuter)#
Si un pipeline comportant 10 étapes échoue à la 9e étape après 4 heures de calculs lourds, vous n'avez pas besoin de relancer tout le DAG depuis le début.
- Cliquez sur la case rouge de la tâche échouée dans la vue Grid.
- Cliquez sur le bouton Clear.
- Choisissez les options de propagation :
- Downstream (En aval - Recommandé) : Relance la tâche sélectionnée ainsi que toutes celles qui en dépendent.
- Upstream (En amont) : Relance aussi les tâches parentes précédentes.
- Recursive : Propage dans les sous-DAGs.
- Airflow efface l'état de la tâche de la base PostgreSQL et la repasse immédiatement en vert clair (running) !
2. Déboguer en ligne de commande avec tasks test#
Tester un pipeline en modifiant le code, en le poussant sur Git et en attendant que le Scheduler le déclenche est un processus laborieux.
Airflow propose une commande extraordinaire pour les développeurs : airflow tasks test.
flowchart TD
subgraph TestCLI["airflow tasks test my_dag my_task 2026-01-15"]
T1["1. Charge le fichier Python du DAG"]
T2["2. Ignore les dépendances amont (exécute la tâche isolée)"]
T3["3. Injecte la date logique passée en paramètre"]
T4["4. Affiche tous les logs et print() directement dans le terminal"]
T5["5. N'enregistre RIEN dans la base de données (0 trace en prod)"]
end
T1 --> T2 --> T3 --> T4 --> T5
style TestCLI fill:#e0f2fe,stroke:#0284c7,stroke-width:2px,color:#0f172a
Comment exécuter un test unitaire ?#
Connectez-vous au conteneur du Scheduler ou du Webserver :
Lancez le test de votre tâche spécifique pour une date historique :
[2026-03-03T10:15:22.412+0000] {taskinstance.py:1571} INFO - Executing <Task(PythonOperator): extraire_donnees> on 2026-02-01 00:00:00+00:00
Extraction réussie. Fichier enregistré sous : /opt/airflow/logs/donnees_brutes.csv
[2026-03-03T10:15:24.110+0000] {taskinstance.py:1400} INFO - Marking task as SUCCESS.
[!TIP] Si votre code Python contient des erreurs de typage ou des exceptions
KeyError, elles s'afficheront instantanément dans votre terminal avec la trace d'appel complète (stack trace), vous faisant gagner un temps précieux lors du développement.
3. Le Backfill : Rattrapage massif d'historique#
Supposons que vous veniez de développer un nouveau pipeline de calcul des commissions commerciales et que la direction souhaite générer les chiffres pour toute l'année 2025.
Au lieu de déclencher le bouton Play 365 fois à la main, utilisez la commande de Backfill :
airflow dags backfill \
--start-date 2025-01-01 \
--end-date 2025-12-31 \
--reset-dagruns \
pipeline_commissions_ventes
Options stratégiques du Backfill#
--start-dateet--end-date: Bornes temporelles (formatYYYY-MM-DD).--reset-dagruns: Écrase et recalcule les exécutions qui existaient déjà sur cette période.--rerun-failed-tasks: Ne relance que les tâches qui étaient en échec dans les exécutions passées.-mou--mark-success: Marque les exécutions historiques comme réussies sans exécuter le code (utile lors de la mise en place d'un nouveau pipeline pour caler la date de départ).
sequenceDiagram
autonumber
participant Dev as Ingénieur Data (CLI)
participant Backfill as Moteur de Backfill
participant Workers as Pool de Workers
Dev->>Backfill: airflow dags backfill --start-date 2025-01-01 --end-date 2025-01-05 my_dag
Backfill->>Backfill: Création des 5 DAG Runs (01/01, 02/01, 03/01, 04/01, 05/01)
par Traitement Parallèle
Backfill->>Workers: Exécution Run 2025-01-01
Backfill->>Workers: Exécution Run 2025-01-02
Backfill->>Workers: Exécution Run 2025-01-03
end
Workers-->>Backfill: Tâches terminées avec succès
Backfill-->>Dev: Backfill achevé sans incident !
4. Automatiser les tests par lot avec un script Python#
Pour valider le bon comportement de votre code sur une série de dates sans polluer la base de production, vous pouvez orchestrer un banc d'essai local en Python :
import os
from datetime import date, timedelta
dag_id = "etl_covid19_daily"
task_id = "transformer_donnees"
date_debut = date(2026, 1, 1)
date_fin = date(2026, 1, 7)
pas = timedelta(days=1)
date_courante = date_debut
while date_courante <= date_fin:
print(f"--- Test unitaire pour la date logique : {date_courante} ---")
commande = f"airflow tasks test {dag_id} {task_id} {date_courante}"
code_retour = os.system(commande)
if code_retour != 0:
print(f"Échec détecté à la date {date_courante} ! Arrêt du banc de test.")
break
date_courante += pas
Vous savez désormais piloter, réparer et rattraper des flux de données avec assurance. Dans la prochaine leçon, nous explorerons les dépendances complexes : branchements conditionnels, trigger rules et capteurs d'événements !
Commentaires