Aller au contenu
Orchestration

Airflow : DAG, tâches et backfill

Un DAG Airflow décrit en Python des tâches et leurs dépendances, sans cycle, avec un planning. On l'utilise pour lancer un pipeline chaque jour, relancer les échecs et rattraper le passé.

  • airflow
  • dag
  • task
  • taskflow
  • retries
  • backfill
  • catchup
  • xcom
  • date logique

Syntaxe

Modèle
from airflow.sdk import dag, task@dag(schedule="0 5 * * *", start_date=…, catchup=False,     default_args={"retries": 2, "retry_delay": timedelta(minutes=10)})def mon_pipeline():    @task    def a(): ...    @task    def b(x): ...    b(a())          # dépendance déduite du passage de valeurmon_pipeline()# Avec des opérateurs classiques : t1 >> t2 >> [t3, t4]

Exemple

Exemple concret
from datetime import datetime, timedeltafrom airflow.sdk import dag, task@dag(    schedule="0 5 * * *",                 # tous les jours à 5 h    start_date=datetime(2026, 10, 1),    catchup=False,    default_args={"retries": 2, "retry_delay": timedelta(minutes=10)},    tags=["ventes"],)def ventes_quotidiennes():    @task    def extraire(ds=None):        # ds : date logique du run (AAAA-MM-JJ), injectée par Airflow        return f"s3://lac/brut/ventes/date={ds}/"    @task    def transformer(chemin: str) -> int:        lignes = nettoyer(chemin)        # fonction métier définie ailleurs        return len(lignes)    @task    def publier(nb_lignes: int):        if nb_lignes == 0:            raise ValueError("Aucune vente : publication bloquée")    publier(transformer(extraire()))ventes_quotidiennes()# Graphe obtenu : extraire → transformer → publier# Si transformer échoue, Airflow le relance deux fois à 10 minutes# d'intervalle, puis marque le run en échec ; publier ne démarre pas.

À retenir

  1. Une valeur retournée par une tâche passe par XCom, stocké dans la base de métadonnées d'Airflow : transmettez un chemin ou un identifiant, jamais un jeu de données entier.
  2. Le backfill relance le DAG sur une plage de dates passées, depuis l'interface ou la ligne de commande (airflow backfill create --dag-id ventes_quotidiennes --from-date 2026-10-01 --to-date 2026-10-07). Il n'est sûr que si chaque tâche est idempotente.
  3. Gardez le code de haut de fichier léger : Airflow relit régulièrement les fichiers de DAG, et une requête ou un import lourd hors des tâches ralentit tout l'ordonnanceur.
  4. Dans les versions plus anciennes, dag et task s'importaient depuis airflow.decorators. Vérifiez aussi la sémantique de la date logique de votre version : selon la table de temps, un run quotidien porte la date de la veille ou celle du déclenchement.
  5. Piège : utiliser datetime.now() dans une tâche. Un backfill du 3 octobre traiterait alors les données d'aujourd'hui ; partez toujours de la date logique du run.

Voir aussi