📊 Apache Airflow – Orchestration, DAGs & Executors

Airflow (v2.0+) a une architecture "désassemblée" (scalable) basée sur 4 composants principaux (plus le dossier de DAGs).

(Utilisateur)
     │
     ▼ (HTTP)
+----------------+
| [ WEBSERVER ]  | (UI / API)
| (Flask/Gunicorn) |
+----------------+
     │ ▲
(Lit/Écrit État) │ (Lit État)
     │ ▼
+----------------+   (Lit/Parse)   +-------------------+
| [ METADATA DB] | ◄--------------- | [ SCHEDULER ]     |
| (Postgres/MySQL) | (Stocke l'État) | (Daemon/Cœur)     |
| (Ex: Runs, Tasks) |                 +-------------------+
+----------------+                       │ ▲
     │                                     │ │ (Parse/Lit)
     │ (Envoie Tâche)                      │ ▼
     ▼                               +-------------+
+----------------+                     | [ DAGs ]    |
| [ EXECUTOR ]   |                     | (Python)    |
| (ex: Celery, K8s)|                     | (.py files) |
| (Le "Muscle")  |                     +-------------+
|   [Worker 1]   |
|   [Worker N]   |
+----------------+

• 1. Webserver (Serveur Web) : (Flask) L'interface utilisateur (UI) pour visualiser les DAGs, monitorer les "Task Instances" (5.1), et déclencher (Trigger) les "DAG Runs" (5.1).
• 2. Scheduler (Planificateur) : (Le "Cœur") Un daemon (service) qui lit (parse) vos fichiers .py (DAGs), vérifie les planifications (schedule) (ex: "cron"), et envoie les tâches prêtes à être exécutées à l'Executor.
• 3. Metadata Database (Base de Données) : (Le "Cerveau") Une base SQL (Postgres/MySQL, jamais SQLite en prod). C'est le SPOF. Elle stocke l'état (state) de tous les DAGs, Tâches, Runs, Logs, XComs, et Connections.
• 4. Executor (Exécuteur) : (Le "Muscle") Définit comment les tâches sont exécutées (Local, Celery, Kubernetes...).
• (5. Dossier de DAGs) : Un dossier (ex: /opt/airflow/dags) (souvent un volume Git-Sync) que le Scheduler et le Webserver lisent.

Cron (voir guide Cron) est un "planificateur" (scheduler). Airflow est un "orchestrateur" (orchestrator).

Le DAG est la définition (le "plan") de votre pipeline. C'est un fichier Python (.py) qui est lu (parsé) par le Scheduler.

Exemple (Fichier .py)

Un DAG est défini en Python. Le code n'est pas exécuté, il est parsé par le Scheduler pour (extraire) la structure (Tâches + Dépendances).

from airflow.decorators import dag
from airflow.operators.bash import BashOperator
from datetime import datetime

# (Arguments par défaut pour le DAG)
default_args = {
    'owner': 'ideo',
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

# (Définition du DAG (le "Graphe"))
@dag(
    dag_id='mon_premier_dag',
    default_args=default_args,
    description='Un DAG simple',
    
    # (Planification : "Cron")
    # (Tous les jours à 2h du matin)
    schedule_interval='0 2 * * *',
    
    # (Date de début (obligatoire))
    start_date=datetime(2025, 1, 1),
    
    # (Ne pas rattraper le passé au 1er lancement)
    catchup=False,
    
    tags=['demo', 'ideo']
)
def ma_fonction_dag():
    
    # (Tâche 1)
    t1 = BashOperator(
        task_id='print_date',
        bash_command='date'
    )
    
    # (Tâche 2)
    t2 = BashOperator(
        task_id='sleep',
        bash_command='sleep 5'
    )
    
    # (Dépendance : t1 PUIS t2)
    t1 >> t2

# (Instanciation du DAG)
ma_fonction_dag()

Un Operator (Opérateur) est la Tâche (Task) atomique. C'est le "nœud" (node) du DAG. C'est le "Quoi" (l'action à exécuter).

# --- 1. Définir la fonction Python ---
# (Peut être dans un autre fichier)
def ma_fonction_python(nom, age):
    print(f"Bonjour, {nom}. Vous avez {age} ans.")
    # (Doit retourner une valeur si xcom_push=True)
    return {"status": 200, "user": nom}

@dag(...)
def mon_dag():

    # --- 2. Définir l'Operator ---
    t_python = PythonOperator(
        task_id='tache_python',
        
        # (La fonction à appeler)
        python_callable=ma_fonction_python,
        
        # (Arguments passés à la fonction)
        op_kwargs={
            "nom": "Alice",
            "age": 30
        }
        
        # (Par défaut, le 'return' est poussé (push)
        #  vers XCom (6.1))
        # do_xcom_push=True 
    )

L'orchestration est définie par les dépendances (flux) entre les tâches (Operators).

Opérateurs "Bitshift" (>>, <<)

La syntaxe "moderne" (Pythonique) pour définir les dépendances (l'ordre).

(t1, t2, t3, t4 sont des Operators)

# --- 1. Séquentiel ---
# (Exécute t1, PUIS t2)
t1 >> t2

# (Équivalent : t2 << t1)
# (Équivalent : t2.set_upstream(t1))

# --- 2. Parallèle (Fan-Out) ---
# (Exécute t1, PUIS (t2 et t3 en parallèle))
t1 >> [t2, t3]

# --- 3. Parallèle (Fan-In) ---
# (Exécute (t1 et t2 en parallèle),
#  PUIS (quand les 2 sont finis) t3)
[t1, t2] >> t3

# --- 4. Chaînage Complexe ---
start >> [t_extract_A, t_extract_B]
t_extract_A >> t_transform_A
t_extract_B >> t_transform_B
[t_transform_A, t_transform_B] >> t_load
t_load >> end

Il faut différencier la Définition (le .py) de l'Exécution (le "Run").

• DAG (Le Plan) : (Fichier mon_dag.py) C'est la Définition (le "plan", la "classe").
• DAG Run (L'Exécution) : (L'"Instance" du plan) C'est 1 exécution du DAG pour un instant T (ex: "L'exécution du 10 Nov 2025").
• Task Instance (L'Instance de Tâche) : (L'"Instance" de la tâche) L'exécution d'1 Tâche (Operator) pour 1 DAG Run spécifique. C'est l'unité la plus petite (ex: "Tâche t1 du Run du 10 Nov").

États (State) d'une Task Instance

C'est ce que l'on voit dans l'UI (les carrés de couleur) :

• none (Gris clair) : (Pas encore démarré)
• queued (Gris foncé) : (Envoyé à l'Executor (Celery)).
• running (Vert clair) : (En cours d'exécution).
• success (Vert foncé) : (Terminé avec succès (exit 0)).
• failed (Rouge) : (Échec (exit non-zéro) après N retries).
• upstream_failed (Orange) : (Ignoré, car une Tâche "parent" (upstream) a échoué).
• skipped (Rose) : (Ignoré (ex: BranchPythonOperator)).
• up_for_retry (Jaune) : (Échec, en attente avant ré-essai).

Un Sensor (Capteur) (une sous-classe d'Operator) est une tâche qui attend (poll) qu'une condition (externe) soit True.

# (Ex: Attendre un fichier S3)
s3_sensor = S3KeySensor(
    task_id='wait_for_s3_file',
    bucket_name='mon-bucket',
    bucket_key='data/input.csv',
    
    mode='poke', # (Défaut)
    poke_interval=60, # (Vérifie toutes les 60 sec)
    timeout=3600 # (Échoue après 1h)
)

s3_sensor = S3KeySensor(
    task_id='wait_for_s3_file',
    ...
    mode='reschedule', # (Libère le slot)
    poke_interval=60,
    timeout=3600
)

Un Hook (Crochet) est une abstraction (wrapper) Python (une classe) qui gère l'interface avec un service externe (BDD, API, Cloud).

Hook vs Operator

Les Operators (2.2) utilisent les Hooks (3.3) (via les Connections (6.2)) pour faire le travail.

(Operator: PostgresOperator)
 "Exécute ce SQL"
   │
   │ (Utilise)
   ▼
(Hook: PostgresHook)
 "Je gère la connexion (lib 'psycopg2'),
  j'ouvre le curseur, j'exécute, je ferme."
   │
   │ (Utilise)
   ▼
(Connection: "my_postgres_db")
 (ID dans l'UI : contient Hôte, Login, Pass)

Usage (Dans un PythonOperator)

On utilise les Hooks directement dans le code Python (PythonOperator) pour de la logique complexe (ex: SELECT -> Transform -> INSERT).

from airflow.providers.postgres.hooks.postgres import PostgresHook

def select_and_insert():
    # 1. (Utilise "my_postgres_db" (6.2))
    pg_hook = PostgresHook(postgres_conn_id='my_postgres_db')
    
    # 2. (Wrapper pour 'psycopg2')
    # (Utilise .get_conn(), .run(), .get_records()...)
    records = pg_hook.get_records("SELECT * FROM table_a")
    
    # (Transformation...)
    
    # 3. Insérer
    pg_hook.insert_rows(table='table_b', rows=...)

Le Scheduler (Planificateur) (airflow scheduler) est le cœur (daemon) d'Airflow.

Rôles (Boucle)

1. 1. Parser (Analyser) les DAGs : (Toutes les N sec) Scanne le dossier dags_folder. Parse tous les fichiers .py (Python) pour détecter les nouveaux DAGs, les Tâches, et les dépendances.
2. 2. Mettre à jour la BDD : Met à jour la Metadata Database (4.3) avec la structure (définition) des DAGs.
3. 3. Planifier (Schedule) : (Toutes les N sec) Interroge la BDD :
   • "Y a-t-il des DAG Runs (schedule_interval) à créer (ex: le cron '0 2 * * *' est arrivé) ?"
   • "Y a-t-il des Task Instances (running) à vérifier (timeout, retry) ?"
   • "Y a-t-il des Task Instances (success) dont les dépendances (A >> B) sont satisfaites ?"
4. 4. Mettre en File (Queue) : Si une Tâche (ex: B) est prête, le Scheduler la marque "queued" (en attente) et l'envoie à l'Executor (5.1) (ex: Celery).

Le Webserver (airflow webserver) est l'Interface Web (GUI) d'Airflow. C'est une application Flask (Python) (généralement servie via Gunicorn en production).

Architecture "Stateless"

L'Interface Web (UI) est (presque) stateless. Elle ne fait rien (pas de planification, pas d'exécution).

Son seul rôle est de lire et écrire dans la Metadata Database (4.3).

Fonctionnalités (UI)

• Vue (DAGs) : Lister, Mettre en Pause (Pause), Déclencher (Trigger DAG).
• Vue (Grid / Tree) : Visualiser l'état (success, failed) des "DAG Runs" et "Task Instances".
• Monitoring : Voir les Logs (récupérés de la BDD ou du worker), voir les Gantt Charts (temps d'exécution).
• Admin (Menu) : Gérer les Connections (6.2), Variables (6.2), Pools (6.3).

La Metadata Database (Base de Données des Métadonnées) est le cœur (cerveau) de l'architecture Airflow. C'est le SPOF (Single Point of Failure).

Stockage (État)

Elle stocke (via SQLAlchemy) tout l'état :

• La liste des DAGs (parsés).
• L'historique de tous les DAG Runs.
• L'état (success, failed...) de toutes les Task Instances.
• Les Variables et Connections (chiffrées).
• Les XComs (données inter-tâches).
• (Si LocalExecutor) Les logs des tâches.

Choix (Dev vs Prod)

• Développement (Défaut) : SQLite (Fichier local airflow.db). (Facile, mais ne supporte pas le parallélisme (LocalExecutor) !).
• Production (Requis) : PostgreSQL ou MySQL. (Nécessaire pour LocalExecutor, CeleryExecutor, KubernetesExecutor).

L'Executor (Exécuteur) est le "moteur" (le "comment"). C'est le composant (configuré dans airflow.cfg) que le Scheduler (4.1) utilise pour exécuter (lancer) physiquement une tâche.

(Scheduler) : "La Tâche A est prête (queued)"
     │
     │ (Ordre : "Exécute Tâche A")
     ▼
[ EXECUTOR ] (ex: Celery)
     │
     │ (Envoie à un Worker)
     ▼
[ Worker (Processus) ]
 (Exécute: 'airflow tasks run ...')

Problème : Comment la Tâche B peut-elle récupérer une valeur (ex: un ID, un nom de fichier) générée par la Tâche A ?

Solution : XCom (Cross-Communication). C'est un système (basé sur la Metadata DB) de "passage" (Push/Pull) de petites données (métadonnées).

(Tâche A)
def f_push(**context):
    # Pousse manuellement (Clé/Valeur)
    context['ti'].xcom_push(key='mon_id', value=12345)

t_push = PythonOperator(
    task_id='t_push',
    python_callable=f_push
)

(Tâche B - Python)
def f_pull(**context):
    # Tire la valeur (key) de la Tâche A (task_ids)
    valeur = context['ti'].xcom_pull(
        key='mon_id',
        task_ids='t_push'
    )
    print(f"Valeur reçue: {valeur}") # 12345

t_pull = PythonOperator(...)

# (Tâche C - Bash)
t_bash = BashOperator(
    task_id='t_bash',
    # (Utilise Jinja pour 'pull' la valeur de retour (default)
    #  de la tâche 't_push')
    
    # --- CORRECTION TEMPLATE DJANGO ---
    bash_command="echo 'Valeur reçue: {{ ti.xcom_pull(task_ids='t_push') }}'"
)

[t_push] >> [t_pull, t_bash]

Ce sont les mécanismes (stockés dans la Metadata DB, gérés via l'UI Admin) pour externaliser la configuration (éviter le "hardcoding" dans les DAGs).

(Usage - Python)
from airflow.models import Variable
seuil = Variable.get("mon_seuil", default_var=50)

(Usage - Jinja)
{{ var.value.mon_seuil }}

(Usage - PythonOperator/Hook)
# (Le Hook lit 'my_postgres_db'
#  automatiquement depuis la BDD)
pg_hook = PostgresHook(postgres_conn_id='my_postgres_db')

Outils pour gérer la concurrence (QoS) et éviter de surcharger les systèmes externes (APIs, BDDs).

t_api = HttpSensor(
    task_id='call_api',
    pool='api_meteo_pool',
    ...
)

t_ml = PythonOperator(
    task_id='train_model',
    queue='heavy_tasks',
    ...
)
t_bash = BashOperator(
    task_id='print_date',
    queue='light_tasks',
    ...
)

catchup (Rattrapage)

Le Piège : Vous créez un DAG (schedule_interval='@daily', start_date='2024-01-01'). Vous le déployez (activez) aujourd'hui (2025-11-13).

Comportement (Défaut : catchup=True) : Airflow voit qu'il a "manqué" ~680 exécutions (depuis 2024-01-01). Il va immédiatement lancer 680 DAG Runs (Backfills), un pour chaque jour manqué. (Spam/DoS).

Solution (Bonne Pratique) : Toujours mettre catchup=False dans le @dag (ou default_args), sauf si vous *voulez* ce comportement.

airflow dags backfill (Rattrapage Manuel)

La commande (CLI) pour exécuter (manuellement) un "rattrapage" (Backfill) sur une période passée, en ignorant les dépendances.

# (Relancer (manuellement) le DAG 'mon_dag'
#  pour la période du 1er au 5 Nov)
$ airflow dags backfill mon_dag \
    --start-date 2025-11-01 \
    --end-date 2025-11-05

Par défaut (trigger_rule='all_success'), une Tâche (B) ne s'exécute que si tous ses parents (A) ont réussi (success).

La trigger_rule (un argument de l'Operator) change ce comportement.

[Tâche A] >> [Tâche C]
[Tâche B] >> [Tâche C]

Ressources officielles pour Apache Airflow :