DAG, tâches et dépendances

Pourquoi un orchestrateur ?

Un pipeline de données réel enchaîne plusieurs étapes : extraire des fichiers ou une API, les nettoyer, les charger dans un entrepôt, puis rafraîchir des tables de reporting. Une ligne cron par script fonctionne au début, puis montre vite ses limites : comment lancer le chargement seulement si l’extraction a réussi ? Comment relancer automatiquement une étape tombée en erreur à cause d’un réseau instable ? Comment retraiter les trois derniers jours ? Où voir l’historique des exécutions ?

Apache Airflow répond à ces questions. Il ne transforme pas les données lui-même : il orchestre des tâches (scripts Python, requêtes SQL, jobs Spark, commandes shell), dans le bon ordre, au bon moment, avec reprises, journaux et interface de suivi. Ce cours utilise Airflow 3 ; les différences importantes avec Airflow 2 sont signalées.

Le vocabulaire : DAG, tâche, exécution

• Un DAG (Directed Acyclic Graph, graphe orienté acyclique) décrit un pipeline : des tâches reliées par des dépendances. « Acyclique » signifie qu’on ne peut jamais revenir à une tâche déjà passée : A → B → A est interdit.
• Une tâche est une unité de travail. Elle est créée à partir d’un opérateur (BashOperator, SQLExecuteQueryOperator…) ou d’une fonction décorée par @task.
• Une exécution de DAG (DAG run) est une instance du pipeline pour une date donnée ; chaque tâche y devient une instance de tâche avec son propre état (queued, running, success, failed, up_for_retry, skipped…).

Le fichier Python d’un DAG est une déclaration : Airflow le relit régulièrement pour connaître la structure du graphe. N’y faites donc aucun appel réseau ni calcul lourd au niveau du module : ce code serait exécuté à chaque analyse du fichier.

Un premier DAG avec TaskFlow

L’API TaskFlow transforme des fonctions Python en tâches :

import pendulum
from airflow.sdk import dag, task


@dag(
    dag_id="ventes_quotidiennes",
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["ventes"],
)
def ventes_quotidiennes():

    @task
    def extraire() -> str:
        # Télécharge l'export du jour et renvoie son chemin (petite valeur)
        return "/data/brut/ventes.csv"

    @task
    def transformer(chemin: str) -> str:
        # Nettoie le fichier et écrit une version propre
        return "/data/propre/ventes.parquet"

    @task
    def charger(chemin: str) -> None:
        print(f"Chargement de {chemin} dans l'entrepôt")

    charger(transformer(extraire()))


ventes_quotidiennes()

L’appel charger(transformer(extraire())) ne lance rien : il construit le graphe extraire → transformer → charger. Les valeurs de retour sont transmises par XCom (cross-communication), un petit stockage tenu dans la base de métadonnées d’Airflow. Passez-y des chemins, des identifiants ou des compteurs, jamais un DataFrame de plusieurs millions de lignes : écrivez les données dans un stockage (fichier, base, stockage objet) et transmettez leur emplacement.

Opérateurs classiques et dépendances explicites

Le même pipeline peut mélanger opérateurs et TaskFlow :

from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.empty import EmptyOperator

with DAG(dag_id="entrepot_nuit", schedule="0 2 * * *",
         start_date=pendulum.datetime(2026, 1, 1, tz="UTC")):
    debut = EmptyOperator(task_id="debut")
    extraire = BashOperator(task_id="extraire", bash_command="python extract.py")
    charger_dw = BashOperator(task_id="charger_dw", bash_command="python load_dw.py")
    charger_lac = BashOperator(task_id="charger_lac", bash_command="python load_lake.py")
    rapport = EmptyOperator(task_id="rapport")

    debut >> extraire >> [charger_dw, charger_lac] >> rapport

a >> b se lit « b s’exécute après a ». Une liste crée une ramification : charger_dw et charger_lac dépendent toutes deux d’extraire et peuvent tourner en parallèle ; rapport attend les deux. Dans Airflow 3, les opérateurs de base (BashOperator, PythonOperator, EmptyOperator, capteurs de fichiers) sont livrés dans le provider standard, d’où l’import airflow.providers.standard.

Comment Airflow choisit l’ordre

Le planificateur (scheduler) applique un principe simple : une tâche devient exécutable quand toutes ses tâches amont sont terminées selon sa règle de déclenchement (par défaut all_success). Parmi les tâches prêtes, plusieurs peuvent être lancées en même temps, dans la limite des ressources (slots, pools, max_active_tasks). Cet ordre est un tri topologique du graphe : il n’existe que parce que le graphe est sans cycle. C’est ce que vous programmerez dans les exercices.

À retenir

• Airflow orchestre : il déclenche, ordonne, relance et trace ; le traitement se fait dans les tâches.
• Un DAG est sans cycle ; le fichier qui le décrit doit rester léger.
• >> et le passage de valeurs TaskFlow créent les dépendances ; XCom ne transporte que de petites valeurs.