Orchestration de Pipelines Data

Ce module présente les outils d'orchestration pour automatiser l'exécution de vos pipelines de données.

Prérequis

Objectifs du module

À la fin de ce notebook, tu seras capable de :

• Comprendre ce qu'est l'orchestration de pipelines
• Utiliser le Planificateur Windows (niveau débutant)
• Configurer des tâches avec Crontab (niveau intermédiaire)
• Comprendre l'architecture d'Apache Airflow
• Créer des DAGs avec Apache Airflow
• Utiliser les différents types d'Operators
• Gérer les dépendances et le passage de données (XCom)
• Configurer les alertes et le monitoring
• Choisir le bon outil selon ton besoin

🎯 L'orchestration dans l'écosystème Data Engineering

Tu as appris à créer des pipelines ETL avec PySpark. Mais comment les automatiser pour qu'ils s'exécutent régulièrement sans intervention manuelle ?

Le problème

Sans orchestration :
┌─────────────────────────────────────────────────────────────┐
│                                                             │
│   😰 "Il faut que je lance mon script tous les jours..."   │
│   😰 "J'ai oublié de lancer le pipeline hier !"            │
│   😰 "Le script B a planté car A n'avait pas fini..."      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Avec orchestration :
┌─────────────────────────────────────────────────────────────┐
│                                                             │
│   ✅ Scripts exécutés automatiquement                       │
│   ✅ Alertes en cas d'échec                                 │
│   ✅ Dépendances respectées (A → B → C)                     │
│   ✅ Logs et monitoring centralisés                         │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Position dans le pipeline Data

┌─────────────────────────────────────────────────────────────────┐
│                     PIPELINE DATA                               │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   Sources        ETL              Destination                   │
│   ────────       ───              ───────────                   │
│                                                                 │
│   APIs     ─┐                 ┌─►  Data Warehouse               │
│   Fichiers ─┼──►  PySpark  ───┼─►  Data Lake                    │
│   BDD      ─┘                 └─►  Dashboard                    │
│                                                                 │
│            ▲                                                    │
│            │                                                    │
│   ┌────────┴────────┐                                           │
│   │  ORCHESTRATION  │  ◄── Quand ? Dans quel ordre ?            │
│   │  (Airflow/Cron) │                                           │
│   └─────────────────┘                                           │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

📊 Comparaison rapide des outils

Niveau 1 : Planificateur Windows

C'est quoi ?

Un outil intégré à Windows pour exécuter des programmes automatiquement.

Comment l'utiliser ?

Ouvrir le planificateur :

Windows + R → Taper 'taskschd.msc' → Entrée

Créer une tâche :

1. Action → Créer une tâche de base
2. Nom : "Mon script quotidien"
3. Déclencheur : Quotidien à 2h du matin
4. Action : Démarrer python.exe avec C:\scripts\mon_script.py
5. Terminer

✅ Voilà ! Votre script s'exécutera automatiquement tous les jours à 2h.

✅ Forces

✅ Très facile - Interface graphique intuitive
✅ Déjà installé - Pas de setup
✅ Parfait pour débuter - Pas de code complexe

❌ Faiblesses

❌ Pas de dépendances - Si tâche A doit finir avant tâche B → compliqué
❌ Monitoring limité - Difficile de voir l'état global
❌ Windows uniquement - Ne fonctionne pas sur Linux

Quand l'utiliser ?

✅ OUI : Vous avez 1-5 scripts simples sur Windows
❌ NON : Vous avez besoin que script B attende script A

Niveau 2 : Crontab (Linux/Mac)

C'est quoi ?

Le planificateur standard sur Linux/Mac.

Syntaxe de base

minute heure jour mois jour_semaine commande

┌───────────── minute (0 - 59)
│ ┌─────────── heure (0 - 23)
│ │ ┌───────── jour du mois (1 - 31)
│ │ │ ┌─────── mois (1 - 12)
│ │ │ │ ┌───── jour de la semaine (0 - 6) (dimanche = 0)
│ │ │ │ │
* * * * * commande

Exemples simples :

# Tous les jours à 2h du matin
0 2 * * * python3 /home/user/script.py

# Toutes les heures
0 * * * * python3 /home/user/hourly.py

# Lundi à vendredi à 9h
0 9 * * 1-5 python3 /home/user/weekday.py

# Toutes les 15 minutes
*/15 * * * * python3 /home/user/check.py

# Le 1er de chaque mois à minuit
0 0 1 * * python3 /home/user/monthly.py

Comment l'utiliser ?

# Éditer votre crontab
crontab -e

# Voir les tâches planifiées
crontab -l

# Ajouter vos lignes
0 2 * * * python3 /home/user/backup.py >> /var/log/backup.log 2>&1

# Sauvegarder et quitter
# ✅ C'est fait !

💡 Astuce : Utilise crontab.guru pour tester tes expressions !

Forces

Universel - Sur TOUS les serveurs Linux
Très léger - Presque pas de ressources
Simple - Une ligne = une tâche
Gratuit - Déjà installé

Faiblesses

Pas de dépendances - Même problème que Windows
Pas de monitoring - Aucune interface
Pas de retry - Si ça échoue, il faut attendre le prochain run
Logs manuels - Il faut les gérer soi-même

Quand l'utiliser ?

OUI : Serveur Linux, 5-15 scripts indépendants
NON : Scripts avec dépendances complexes

Niveau 3 : Apache Airflow

C'est quoi ?

Un orchestrateur professionnel pour gérer des workflows complexes, créé par Airbnb en 2014 et devenu projet Apache en 2016.

┌─────────────────────────────────────────────────────────────────┐
│                    APACHE AIRFLOW                               │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   "Airflow is a platform to programmatically author,           │
│    schedule, and monitor workflows."                            │
│                                                                 │
│   • Créé par Airbnb (2014)                                      │
│   • Apache Top-Level Project (2019)                             │
│   • 30,000+ GitHub stars                                        │
│   • Utilisé par : Airbnb, Netflix, Spotify, Twitter, Adobe...   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

La grande différence avec Cron :

# ❌ Avec Cron (problème) :
0 2 * * * python extract.py
30 2 * * * python transform.py  # Et si extract prend plus de 30min ?
0 3 * * * python load.py        # Et si transform a planté ?

# ✅ Avec Airflow (solution) :
extract >> transform >> load  # transform ATTEND que extract soit terminé !

Architecture d'Airflow

┌─────────────────────────────────────────────────────────────────────────────┐
│                        ARCHITECTURE AIRFLOW                                 │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│   ┌─────────────────┐         ┌─────────────────┐         ┌─────────────┐  │
│   │                 │         │                 │         │             │  │
│   │   WEB SERVER    │◄───────►│    SCHEDULER    │◄───────►│  EXECUTOR   │  │
│   │   (Flask UI)    │         │  (Orchestrate)  │         │  (Workers)  │  │
│   │                 │         │                 │         │             │  │
│   └────────┬────────┘         └────────┬────────┘         └──────┬──────┘  │
│            │                           │                         │         │
│            │                           ▼                         │         │
│            │                  ┌─────────────────┐                │         │
│            │                  │                 │                │         │
│            └─────────────────►│    METADATA     │◄───────────────┘         │
│                               │    DATABASE     │                          │
│                               │  (PostgreSQL)   │                          │
│                               │                 │                          │
│                               └─────────────────┘                          │
│                                        ▲                                   │
│                                        │                                   │
│                               ┌────────┴────────┐                          │
│                               │                 │                          │
│                               │    DAG FILES    │                          │
│                               │   (Python .py)  │                          │
│                               │                 │                          │
│                               └─────────────────┘                          │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

Composants clés

Types d'Executors

Concepts clés d'Airflow

1️⃣ DAG (Directed Acyclic Graph)

Un DAG est un graphe de tâches sans cycle : les données vont toujours dans une direction.

     ┌─────────┐
     │ Extract │
     └────┬────┘
          │
          ▼
     ┌─────────┐
     │Transform│
     └────┬────┘
          │
    ┌─────┴─────┐
    ▼           ▼
┌───────┐  ┌────────┐
│Load DW│  │Load S3 │
└───────┘  └────────┘

2️⃣ Task

Une Task est une unité de travail dans un DAG (une étape).

3️⃣ Operator

Un Operator définit ce que fait une tâche.

4️⃣ Task Instance

Une Task Instance = une exécution spécifique d'une Task à une date donnée.

5️⃣ DAG Run

Un DAG Run = une exécution complète du DAG.

DAG: etl_pipeline
├── DAG Run 2024-01-15 ✅
│   ├── extract (Task Instance) ✅
│   ├── transform (Task Instance) ✅
│   └── load (Task Instance) ✅
│
├── DAG Run 2024-01-16 ⏳
│   ├── extract (Task Instance) ✅
│   ├── transform (Task Instance) ⏳ running
│   └── load (Task Instance) ⏸️ waiting

Installation locale d'Airflow

# Créer un environnement virtuel
python -m venv airflow_venv
source airflow_venv/bin/activate  # Linux/Mac
# ou : airflow_venv\Scripts\activate  # Windows

# Définir le home Airflow
export AIRFLOW_HOME=~/airflow

# Installer Airflow (version contrainte pour compatibilité)
AIRFLOW_VERSION=2.8.1
PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"
CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

# Initialiser la base de données
airflow db init

# Créer un utilisateur admin
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com \
    --password admin

# Lancer le webserver (Terminal 1)
airflow webserver --port 8080

# Lancer le scheduler (Terminal 2)
airflow scheduler

👉 Accéder : http://localhost:8080 (login: admin / admin)

Ton premier DAG

Créer le fichier ~/airflow/dags/mon_premier_dag.py :

# ~/airflow/dags/mon_premier_dag.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

# Arguments par défaut pour toutes les tâches
default_args = {
    'owner': 'data_engineer',
    'depends_on_past': False,
    'email': ['data-team@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

# Définir le DAG
dag = DAG(
    dag_id='mon_premier_dag',
    default_args=default_args,
    description='Mon premier pipeline ETL',
    schedule_interval='@daily',  # Exécution quotidienne
    start_date=datetime(2024, 1, 1),
    catchup=False,  # Ne pas exécuter les runs passés
    tags=['etl', 'tutorial'],
)

# Fonctions Python
def extract():
    """Extraire les données"""
    print("📥 Extraction des données depuis l'API...")
    # Simuler extraction
    data = {'records': 1000, 'source': 'api'}
    return data  # Retourné via XCom

def transform(**context):
    """Transformer les données"""
    # Récupérer les données de extract via XCom
    ti = context['ti']
    data = ti.xcom_pull(task_ids='extract')
    print(f"🔄 Transformation de {data['records']} enregistrements")
    return {'records': data['records'], 'cleaned': True}

def load(**context):
    """Charger les données"""
    ti = context['ti']
    data = ti.xcom_pull(task_ids='transform')
    print(f"💾 Chargement de {data['records']} enregistrements nettoyés")

# Créer les tâches
task_start = BashOperator(
    task_id='start',
    bash_command='echo "🚀 Démarrage du pipeline à $(date)"',
    dag=dag,
)

task_extract = PythonOperator(
    task_id='extract',
    python_callable=extract,
    dag=dag,
)

task_transform = PythonOperator(
    task_id='transform',
    python_callable=transform,
    dag=dag,
)

task_load = PythonOperator(
    task_id='load',
    python_callable=load,
    dag=dag,
)

task_end = BashOperator(
    task_id='end',
    bash_command='echo "✅ Pipeline terminé avec succès !"',
    dag=dag,
)

# Définir les dépendances
task_start >> task_extract >> task_transform >> task_load >> task_end

# Équivalent à :
# task_start.set_downstream(task_extract)
# task_extract.set_downstream(task_transform)
# etc.

Paramètres importants du DAG

Schedule Interval (fréquence d'exécution)

Catchup

# catchup=True (défaut) :
# Si start_date=2024-01-01 et on est le 2024-01-10,
# Airflow va exécuter les 10 DAG Runs manqués !

# catchup=False :
# Exécute seulement à partir de maintenant

Default Args importants

default_args = {
    'owner': 'data_team',           # Propriétaire
    'depends_on_past': False,       # Dépend du run précédent ?
    'email_on_failure': True,       # Email si échec
    'retries': 3,                   # Nombre de retry
    'retry_delay': timedelta(minutes=5),  # Délai entre retries
    'execution_timeout': timedelta(hours=1),  # Timeout
    'sla': timedelta(hours=2),      # SLA (alerte si dépassé)
}

Les Operators les plus utilisés

PythonOperator

from airflow.operators.python import PythonOperator

def my_function(name, **context):
    print(f"Hello {name}!")
    print(f"Execution date: {context['ds']}")
    return "success"

task = PythonOperator(
    task_id='python_task',
    python_callable=my_function,
    op_kwargs={'name': 'World'},  # Arguments de la fonction
)

BashOperator

from airflow.operators.bash import BashOperator

task = BashOperator(
    task_id='bash_task',
    bash_command='echo "Date: {{ ds }}" && python /scripts/etl.py',
)

PostgresOperator

from airflow.providers.postgres.operators.postgres import PostgresOperator

task = PostgresOperator(
    task_id='create_table',
    postgres_conn_id='my_postgres',  # Connexion définie dans UI
    sql="""
        CREATE TABLE IF NOT EXISTS users (
            id SERIAL PRIMARY KEY,
            name VARCHAR(100)
        );
    """,
)

EmailOperator

from airflow.operators.email import EmailOperator

task = EmailOperator(
    task_id='send_report',
    to='team@company.com',
    subject='Pipeline Report - {{ ds }}',
    html_content='<h1>Pipeline completed!</h1>',
)

EmptyOperator (anciennement DummyOperator)

from airflow.operators.empty import EmptyOperator

# Utile pour créer des points de jonction
start = EmptyOperator(task_id='start')
end = EmptyOperator(task_id='end')

XCom — Passer des données entre tâches

XCom (Cross-Communication) permet de partager des données entre tâches.

┌──────────┐         XCom          ┌───────────┐
│  Task A  │ ─────── data ───────► │   Task B  │
│  return  │                       │ xcom_pull │
└──────────┘                       └───────────┘

Méthode 1 : Return (automatique)

# XCom avec return (automatique)

def extract():
    data = {'records': 1000, 'status': 'ok'}
    return data  # Automatiquement stocké dans XCom

def transform(**context):
    # Récupérer via ti (task instance)
    ti = context['ti']
    data = ti.xcom_pull(task_ids='extract')
    print(f"Reçu: {data}")
    return {'records': data['records'], 'transformed': True}

Méthode 2 : Push/Pull explicite

def task_a(**context):
    # Push explicite avec une clé
    context['ti'].xcom_push(key='my_data', value={'count': 42})
    context['ti'].xcom_push(key='status', value='success')

def task_b(**context):
    # Pull avec la clé
    data = context['ti'].xcom_pull(task_ids='task_a', key='my_data')
    status = context['ti'].xcom_pull(task_ids='task_a', key='status')

⚠️ Limites de XCom

# ❌ Mauvaise pratique
def extract():
    df = pd.read_csv('big_file.csv')  # 1GB
    return df.to_dict()  # ❌ Trop gros pour XCom !

# ✅ Bonne pratique
def extract():
    df = pd.read_csv('big_file.csv')
    path = '/data/output/extract_2024-01-15.parquet'
    df.to_parquet(path)
    return path  # ✅ Passer le chemin, pas les données

TaskFlow API (Airflow 2.0+)

Syntaxe moderne et plus Pythonic avec des décorateurs :

# TaskFlow API - Syntaxe moderne (Airflow 2.0+)

from airflow.decorators import dag, task
from datetime import datetime

@dag(
    dag_id='etl_taskflow',
    schedule='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'taskflow'],
)
def etl_pipeline():
    
    @task()
    def extract():
        """Extraire les données"""
        return {'records': 1000, 'source': 'api'}
    
    @task()
    def transform(data: dict):
        """Transformer les données"""
        return {
            'records': data['records'],
            'cleaned': True
        }
    
    @task()
    def load(data: dict):
        """Charger les données"""
        print(f"Loading {data['records']} records")
    
    # Définir le flow - XCom automatique !
    raw_data = extract()
    clean_data = transform(raw_data)
    load(clean_data)

# Instancier le DAG
etl_pipeline()

Avantages TaskFlow API

Sensors — Attendre une condition

Les Sensors attendent qu'une condition soit remplie avant de continuer.

┌──────────────────────────────────────────────────────────────┐
│                                                              │
│   FileSensor          S3KeySensor         HttpSensor         │
│   "Fichier existe?"   "Fichier sur S3?"   "API disponible?"  │
│                                                              │
│        ⏳                   ⏳                   ⏳            │
│        │                    │                    │           │
│        ▼                    ▼                    ▼           │
│       ✅                   ✅                   ✅           │
│   Continuer            Continuer            Continuer        │
│                                                              │
└──────────────────────────────────────────────────────────────┘

FileSensor

# FileSensor - Attendre qu'un fichier existe

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id='wait_for_file',
    filepath='/data/input/daily_export.csv',
    poke_interval=60,      # Vérifier toutes les 60 secondes
    timeout=3600,          # Timeout après 1 heure
    mode='poke',           # poke ou reschedule
)

# Le pipeline attend le fichier avant de continuer
wait_for_file >> process_file

Autres Sensors utiles

# S3KeySensor - Attendre un fichier sur S3
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_s3 = S3KeySensor(
    task_id='wait_for_s3',
    bucket_name='my-bucket',
    bucket_key='data/{{ ds }}/export.csv',
    aws_conn_id='aws_default',
)

# HttpSensor - Attendre qu'une API réponde
from airflow.providers.http.sensors.http import HttpSensor

wait_api = HttpSensor(
    task_id='wait_for_api',
    http_conn_id='api_connection',
    endpoint='health',
    response_check=lambda response: response.status_code == 200,
)

# ExternalTaskSensor - Attendre un autre DAG
from airflow.sensors.external_task import ExternalTaskSensor

wait_other_dag = ExternalTaskSensor(
    task_id='wait_upstream',
    external_dag_id='upstream_dag',
    external_task_id='final_task',
)

Mode poke vs reschedule

Branching — Logique conditionnelle

Exécuter différentes tâches selon une condition.

                    ┌──────────┐
                    │  Check   │
                    │ Condition│
                    └────┬─────┘
                         │
              ┌──────────┼──────────┐
              │          │          │
              ▼          ▼          ▼
         ┌────────┐ ┌────────┐ ┌────────┐
         │ Path A │ │ Path B │ │ Path C │
         └────────┘ └────────┘ └────────┘

# Branching - Exécution conditionnelle

from airflow.operators.python import BranchPythonOperator
from airflow.operators.empty import EmptyOperator

def choose_branch(**context):
    """Décider quelle branche exécuter"""
    # Exemple : vérifier le jour de la semaine
    day = context['ds_nodash']  # Format YYYYMMDD
    
    # Logique métier
    if int(day) % 2 == 0:
        return 'process_even'  # Retourner le task_id à exécuter
    else:
        return 'process_odd'

branch = BranchPythonOperator(
    task_id='branch',
    python_callable=choose_branch,
)

process_even = EmptyOperator(task_id='process_even')
process_odd = EmptyOperator(task_id='process_odd')
end = EmptyOperator(task_id='end', trigger_rule='none_failed_min_one_success')

# Définir le flow
branch >> [process_even, process_odd] >> end

Trigger Rules

Contrôler quand une tâche s'exécute en fonction du statut des tâches parentes.

from airflow.utils.trigger_rule import TriggerRule

# Tâche de notification en cas d'échec
notify_failure = EmailOperator(
    task_id='notify_failure',
    to='team@company.com',
    subject='Pipeline Failed!',
    html_content='...',
    trigger_rule=TriggerRule.ONE_FAILED,  # Exécute si un parent échoue
)

# Tâche finale qui s'exécute toujours
cleanup = BashOperator(
    task_id='cleanup',
    bash_command='rm -rf /tmp/data/*',
    trigger_rule=TriggerRule.ALL_DONE,  # Toujours exécuter
)

Connections et Variables

Connections

Stocker les informations de connexion aux systèmes externes.

Dans l'UI : Admin → Connections → +

┌─────────────────────────────────────────────────────────────┐
│  Add Connection                                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  Connection Id:   │ my_postgres                 │           │
│  Connection Type: │ Postgres           ▼        │           │
│  Host:            │ localhost                   │           │
│  Schema:          │ mydb                        │           │
│  Login:           │ admin                       │           │
│  Password:        │ ••••••••                    │           │
│  Port:            │ 5432                        │           │
│                                                             │
│                              [ Save ]                       │
└─────────────────────────────────────────────────────────────┘

Utilisation dans le code :

from airflow.hooks.postgres_hook import PostgresHook

def query_postgres():
    hook = PostgresHook(postgres_conn_id='my_postgres')
    df = hook.get_pandas_df('SELECT * FROM users')
    return df

Variables

Stocker des configurations réutilisables.

Dans l'UI : Admin → Variables → +

from airflow.models import Variable

# Récupérer une variable
api_key = Variable.get('API_KEY')

# Variable JSON
config = Variable.get('pipeline_config', deserialize_json=True)
# config = {'batch_size': 1000, 'env': 'prod'}

# Dans un template Jinja
# {{ var.value.API_KEY }}
# {{ var.json.pipeline_config.batch_size }}

Monitoring et Alertes

Interface Web

┌─────────────────────────────────────────────────────────────────────────────┐
│  Airflow - DAGs                                                             │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  DAG                    Schedule    Owner    Runs   Recent Tasks            │
│  ─────────────────────  ─────────   ─────    ────   ────────────            │
│  ▶ etl_pipeline         @daily      team     125    ✅✅✅✅✅                │
│  ▶ data_quality_check   @hourly     team     560    ✅✅✅❌✅                │
│  ▶ weekly_report        @weekly     team     52     ✅✅✅✅✅                │
│  ⏸ maintenance          None        admin    3      ✅✅✅                   │
│                                                                             │
│  ✅ Success  ❌ Failed  ⏳ Running  ⏸ Paused                                 │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

Vues disponibles

Configurer les alertes email

# airflow.cfg
[smtp]
smtp_host = smtp.gmail.com
smtp_port = 587
smtp_user = airflow@company.com
smtp_password = your_password
smtp_mail_from = airflow@company.com

# Dans le DAG
default_args = {
    'email': ['team@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
}

Alertes Slack

from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator

def alert_slack_on_failure(context):
    """Callback en cas d'échec"""
    slack_msg = f"""
        :red_circle: Task Failed!
        *DAG*: {context['dag'].dag_id}
        *Task*: {context['task'].task_id}
        *Execution Time*: {context['execution_date']}
    """
    return SlackWebhookOperator(
        task_id='slack_alert',
        slack_webhook_conn_id='slack_webhook',
        message=slack_msg,
    ).execute(context)

# Utiliser le callback
default_args = {
    'on_failure_callback': alert_slack_on_failure,
}

Bonnes pratiques Airflow

1. Structure des DAGs

airflow/
├── dags/
│   ├── __init__.py
│   ├── etl/
│   │   ├── __init__.py
│   │   ├── daily_etl.py
│   │   └── weekly_report.py
│   ├── utils/
│   │   ├── __init__.py
│   │   └── helpers.py
│   └── config/
│       └── settings.py
├── plugins/
│   └── custom_operators/
└── tests/
    └── test_dags.py

2. Idempotence

# ❌ Non idempotent - accumule des données
def bad_load():
    db.execute("INSERT INTO table VALUES (...)")

# ✅ Idempotent - même résultat si relancé
def good_load():
    db.execute("DELETE FROM table WHERE date = '{{ ds }}'")
    db.execute("INSERT INTO table SELECT ... WHERE date = '{{ ds }}'")

3. Atomicité des tâches

# ❌ Tâche trop grosse
def do_everything():
    extract()
    transform()
    load()

# ✅ Tâches atomiques
extract >> transform >> load

4. Ne pas mettre de logique dans le DAG

# ❌ Code exécuté à chaque parsing
data = fetch_from_api()  # Appelé toutes les 30s !

# ✅ Logique dans les tasks
@task
def fetch_data():
    return fetch_from_api()

5. Tester les DAGs

# tests/test_dags.py
import pytest
from airflow.models import DagBag

def test_dag_loaded():
    dag_bag = DagBag()
    assert len(dag_bag.import_errors) == 0

def test_dag_structure():
    dag_bag = DagBag()
    dag = dag_bag.get_dag('etl_pipeline')
    assert dag is not None
    assert len(dag.tasks) == 5

✅ Forces d'Airflow

Gestion des dépendances - A >> B = B attend A
Retry automatique - Réessaie en cas d'échec
Interface web - Visualisation complète
Monitoring - Logs centralisés
Alertes - Email/Slack en cas d'échec
Scalable - Gère 100+ pipelines
Extensible - Custom operators, hooks
Communauté - Très active, beaucoup de providers

❌ Faiblesses d'Airflow

Complexe - Courbe d'apprentissage
Ressources - Besoin de 4-8 GB RAM
Overkill - Pour 1-3 scripts simples
Setup - Installation et configuration
Pas pour le streaming - Batch only (utiliser Kafka)

Quand utiliser Airflow ?

OUI : 10+ pipelines, dépendances complexes, production
NON : 1-5 scripts simples sans dépendances

Tableau de décision

Quel outil choisir ?

Progression naturelle

1. Débutez avec Task Scheduler ou Cron
2. Quand vous avez 5+ scripts → Pensez à migrer
3. Quand vous avez des dépendances → Migrez vers Airflow

Résumé

Points clés

Windows Task Scheduler

• Pour qui : Débutants sur Windows
• Force : Très facile (GUI)
• Faiblesse : Pas de dépendances
• Limite : 5 scripts max

Crontab

• Pour qui : Utilisateurs Linux
• Force : Universel, léger
• Faiblesse : Pas de monitoring
• Limite : 15 scripts max

Airflow

• Pour qui : Production, équipes data
• Force : Dépendances, monitoring, scalable
• Faiblesse : Complexe, ressources
• Limite : Aucune (scalable)

Concepts Airflow à retenir

Ressources

• Crontab : crontab.guru (tester expressions)
• Airflow : airflow.apache.org
• Airflow Tutorial : Documentation officielle

Quiz Final

❓ Q1. Vous avez 2 scripts Python à exécuter tous les jours sur Windows. Quel outil ?

a) Airflow
b) Crontab
c) Windows Task Scheduler
d) Kubernetes

❓ Q2. Expression crontab pour "tous les lundis à 9h" ?

a) 0 9 * * 1
b) 9 0 * * 1
c) * 9 * * 1
d) 0 9 1 * *

❓ Q3. Que signifie DAG dans Airflow ?

a) Data Analysis Graph
b) Directed Acyclic Graph
c) Dynamic Airflow Generator
d) Database Access Gateway

❓ Q4. Dans Airflow, que signifie task_a >> task_b ?

a) task_a et task_b en parallèle
b) task_b attend que task_a finisse
c) task_a attend task_b
d) Erreur de syntaxe

❓ Q5. Quel composant Airflow décide quand exécuter les tâches ?

a) Web Server
b) Executor
c) Scheduler
d) Metadata DB

❓ Q6. Comment passer des données entre tâches Airflow ?

a) Variables globales
b) XCom
c) Fichiers partagés uniquement
d) Pas possible

❓ Q7. Qu'est-ce qu'un Sensor dans Airflow ?

a) Un outil de monitoring
b) Un Operator qui attend une condition
c) Un type de DAG
d) Un système d'alerte

❓ Q8. Quand NE PAS utiliser Airflow ?

a) 10+ pipelines complexes
b) 1-3 scripts simples
c) Production critique
d) Besoin de monitoring

❓ Q9. Quel est l'avantage de la TaskFlow API ?

a) Plus performant
b) XCom automatique et code plus lisible
c) Compatible Python 2
d) Pas besoin de Scheduler

❓ Q10. Vous avez 15 scripts avec des dépendances. Quel outil ?

a) Windows Task Scheduler
b) Crontab
c) Apache Airflow
d) Aucun, faire manuellement

📊 Votre score

• 10/10 : 🏆 Expert orchestration !
• 8-9/10 : 🌟 Très bien !
• 6-7/10 : 💪 Bon début !
• < 6/10 : 📚 Relisez le notebook

📚 Ressources

Outils

• Crontab Guru — Tester et générer des expressions cron
• Apache Airflow — Documentation officielle
• Airflow Tutorial
• Astronomer — Guides et bonnes pratiques

Alternatives à Airflow

Cloud managed

🎉 Fin du niveau débutant !

Tu as terminé le parcours Data Engineering - From Zero to Hero niveau débutant ! 🎉

Récapitulatif des modules

Module BONUS disponible

👉 Parfait pour exposer les résultats de tes pipelines via API !

➡️ Prochaine étape : Niveau Intermédiaire

Tu es maintenant prêt pour le niveau intermédiaire qui couvrira :

🚀 Bravo ! Tu as maintenant toutes les bases pour construire des pipelines de données !