Authoring Airflow DAGs

Rédaction de DAGs Airflow

Quand l'utiliser

• Créer ou refactoriser des DAGs et tâches Airflow.
• Configurer les ordonnancements, rattrapages/backfills, tentatives et SLA.
• Passer des données entre tâches (XCom) ou utiliser des connexions/variables.
• NE PAS l'utiliser pour diagnostiquer un DAG en panne (utiliser debugging-airflow-pipelines).

Workflow

- [ ] Rendre chaque tâche idempotente et paramétrée par l'intervalle de données
- [ ] Garder le code coûteux/lourd à l'intérieur des tâches, pas au niveau supérieur du module
- [ ] Définir délibérément l'ordonnancement et le rattrapage
- [ ] Configurer les tentatives, le délai de tentative et les SLA
- [ ] Câbler les dépendances via les valeurs de retour TaskFlow ou les opérateurs >>

1. Tâches idempotentes — une tâche pour l'intervalle 2026-01-15 doit produire le même résultat qu'elle soit exécutée une fois ou relancée. Utiliser l'intervalle de données, pas datetime.now().
2. Pas de code lourd au niveau supérieur — le scheduler analyse fréquemment chaque fichier DAG ; des appels de base de données, des appels API ou des imports volumineux au niveau module ralentissent l'ordonnancement et peuvent casser l'analyse. Les placer dans les tâches.
3. Ordonnancement et rattrapage délibérés — catchup=True remplit chaque intervalle manqué depuis start_date ; par défaut à False sauf si souhaité.
4. Tentatives et SLA — les échecs transitoires sont normaux ; définir retries et retry_delay ; utiliser des SLA/alertes pour les retards.

Patterns

DAG TaskFlow, idempotent et proprement câblé :

from airflow.decorators import dag, task
import pendulum

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=5)},
    tags=["commandes"],
)
def pipeline_commandes():

    @task
    def extraire(data_interval_start=None):
        # Utiliser l'intervalle, pas now(), pour que les relances soient déterministes.
        return fetch_orders(day=data_interval_start.date())

    @task
    def charger(lignes):
        # Supprimer-insérer la partition -> idempotent en cas de tentative.
        overwrite_partition("fct_orders", lignes)

    charger(extraire())

pipeline_commandes()

Passer de petites données via XCom (valeurs de retour) ; passer de gros volumes via le stockage — écrire vers S3/GCS/warehouse et passer le chemin/clé, jamais de mégaoctets via XCom.

Utiliser les connexions/variables pour les secrets et la configuration (BaseHook.get_connection, Variable.get), jamais de credentials codés en dur.

Pièges courants

• Appels API/DB au niveau supérieur ou imports lourds — ralentissent le scheduler et peuvent faire échouer l'analyse des DAGs sur tout le déploiement.
• datetime.now() à l'intérieur des tâches — casse l'idempotence et les backfills ; utiliser data_interval_start/_end.
• catchup=True non intentionnel — inonde le cluster de runs historiques lors du premier déploiement.
• Gros volumes via XCom — gonfle la base de données de métadonnées ; passer des références.
• start_date dynamique (ex. days_ago) — rend les ordonnancements non déterministes ; utiliser un timestamp fixe.
• Une seule tâche monstrueuse — diviser extract/transform/load pour que les tentatives soient granulaires.

Références

• Référence sur l'ordonnancement, le rattrapage et les backfills