🎼 Advanced Orchestration pour Data Engineers

Bienvenue dans ce module avancé où tu vas maîtriser l'orchestration de pipelines data à l'échelle. Tu apprendras à déployer Airflow sur Kubernetes, à exploiter la TaskFlow API, à comparer les orchestrateurs modernes, et à mettre en place le data lineage avec OpenLineage — des compétences essentielles pour un Data Engineer Senior !

Prérequis

🎯 Objectifs du module

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

• Déployer et configurer Airflow sur Kubernetes
• Utiliser le KubernetesExecutor et le KubernetesPodOperator
• Écrire des DAGs modernes avec la TaskFlow API
• Implémenter le Dynamic Task Mapping
• Comparer et choisir entre Airflow, Dagster et Prefect
• Mettre en place le data lineage avec OpenLineage
• Utiliser Astronomer (Astro CLI et Astro SDK) pour industrialiser

Rappel : Ce qu'on a vu vs Ce qu'on va approfondir

Schéma : De l'orchestration basique à l'orchestration avancée

M12 Orchestration Basics              M28 Advanced Orchestration (ce module)
┌─────────────────────────┐           ┌─────────────────────────────────────┐
│ • Cron / Task Scheduler │           │ • Airflow sur Kubernetes            │
│ • DAGs basics           │           │ • KubernetesExecutor                │
│ • Operators simples     │  ──────▶  │ • TaskFlow API                      │
│ • XCom manuel           │           │ • Dynamic Task Mapping              │
│ • SequentialExecutor    │           │ • Dagster, Prefect (comparatif)     │
└─────────────────────────┘           │ • OpenLineage (lineage)             │
                                      │ • Astronomer (enterprise)           │
                                      └─────────────────────────────────────┘

💡 Ce module est orienté "production à l'échelle" — tu vas apprendre à faire tourner des centaines de DAGs avec des milliers de tâches sur Kubernetes.

ℹ️ Le savais-tu ?

Apache Airflow a été créé par Airbnb en 2014 pour orchestrer leurs pipelines data. Il a été donné à la Apache Foundation en 2016.

Aujourd'hui, Airflow est utilisé par des milliers d'entreprises dont Uber, Lyft, Twitter, Slack, Adobe, et bien d'autres.

Astronomer, fondé en 2018, est devenu le leader des solutions Airflow managées, levant plus de 200 millions de dollars et étant le principal contributeur au projet open-source Airflow.

📖 History of Apache Airflow

1. Airflow sur Kubernetes

Déployer Airflow sur Kubernetes permet de bénéficier de l'élasticité, de l'isolation et de la scalabilité native de K8s.

Pourquoi Airflow sur K8s ?

Architecture Airflow sur Kubernetes

┌─────────────────────────────────────────────────────────────────────────────┐
│                        AIRFLOW ON KUBERNETES                                │
│                                                                             │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      Namespace: airflow                              │   │
│  │                                                                      │   │
│  │   ┌─────────────┐   ┌─────────────┐   ┌─────────────────────────┐   │   │
│  │   │  Scheduler  │   │  Webserver  │   │   Triggerer (Airflow 2) │   │   │
│  │   │   (Pod)     │   │   (Pod)     │   │       (Pod)             │   │   │
│  │   └──────┬──────┘   └─────────────┘   └─────────────────────────┘   │   │
│  │          │                                                           │   │
│  │          │ Crée des pods pour chaque tâche                          │   │
│  │          ▼                                                           │   │
│  │   ┌─────────────────────────────────────────────────────────────┐   │   │
│  │   │              Worker Pods (éphémères)                         │   │   │
│  │   │   ┌───────┐  ┌───────┐  ┌───────┐  ┌───────┐  ┌───────┐    │   │   │
│  │   │   │Task 1 │  │Task 2 │  │Task 3 │  │Task 4 │  │Task 5 │    │   │   │
│  │   │   │ Pod   │  │ Pod   │  │ Pod   │  │ Pod   │  │ Pod   │    │   │   │
│  │   │   └───────┘  └───────┘  └───────┘  └───────┘  └───────┘    │   │   │
│  │   └─────────────────────────────────────────────────────────────┘   │   │
│  │                                                                      │   │
│  │   ┌─────────────────┐   ┌─────────────────────────────────────┐     │   │
│  │   │   PostgreSQL    │   │            DAGs (PVC/Git-Sync)      │     │   │
│  │   │  (Metadata DB)  │   │                                     │     │   │
│  │   └─────────────────┘   └─────────────────────────────────────┘     │   │
│  └─────────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────────┘

Les Executors Airflow

Installation avec le Helm Chart Officiel

# Ajouter le repo Helm officiel Airflow
helm repo add apache-airflow https://airflow.apache.org
helm repo update

# Créer le namespace
kubectl create namespace airflow

# Installer Airflow avec KubernetesExecutor
helm install airflow apache-airflow/airflow \
  --namespace airflow \
  --set executor=KubernetesExecutor \
  --set webserver.defaultUser.password=admin \
  --set dags.persistence.enabled=true \
  --set dags.gitSync.enabled=true \
  --set dags.gitSync.repo=https://github.com/myorg/airflow-dags.git \
  --set dags.gitSync.branch=main \
  --set dags.gitSync.subPath=dags

# Vérifier l'installation
kubectl get pods -n airflow

# Accéder au webserver
kubectl port-forward svc/airflow-webserver -n airflow 8080:8080
# http://localhost:8080 (admin / admin)

Configuration values.yaml avancée

# values-production.yaml
executor: KubernetesExecutor

# Webserver
webserver:
  replicas: 2
  resources:
    requests:
      cpu: "500m"
      memory: "1Gi"
    limits:
      cpu: "1000m"
      memory: "2Gi"

# Scheduler
scheduler:
  replicas: 2    # HA Scheduler (Airflow 2.0+)
  resources:
    requests:
      cpu: "500m"
      memory: "1Gi"

# Configuration KubernetesExecutor
config:
  kubernetes:
    # Namespace pour les worker pods
    namespace: airflow
    # Supprimer les pods après exécution
    delete_worker_pods: "True"
    delete_worker_pods_on_failure: "False"  # Garder pour debug
    # Image par défaut pour les workers
    worker_container_repository: apache/airflow
    worker_container_tag: 2.8.0-python3.11

# Git-Sync pour les DAGs
dags:
  persistence:
    enabled: false
  gitSync:
    enabled: true
    repo: git@github.com:myorg/airflow-dags.git
    branch: main
    subPath: dags
    sshKeySecret: airflow-git-ssh-key
    wait: 60  # Sync toutes les 60 secondes

# Logs dans un stockage externe
logs:
  persistence:
    enabled: true
    size: 10Gi

# PostgreSQL (ou utiliser un service externe)
postgresql:
  enabled: true
  auth:
    postgresPassword: airflow
    username: airflow
    password: airflow
    database: airflow

KubernetesExecutor : Comment ça marche

1. DAG est schedulé
       │
       ▼
2. Scheduler parse le DAG et identifie les tâches à exécuter
       │
       ▼
3. Pour chaque tâche, le Scheduler crée un Pod K8s
       │
   ┌───┴───────────────────────────────────────┐
   │  Pod Spec généré automatiquement :        │
   │  - Image: airflow (ou custom)             │
   │  - Command: airflow tasks run ...         │
   │  - Env: connexions, variables             │
   │  - Resources: requests/limits             │
   └───────────────────────────────────────────┘
       │
       ▼
4. K8s schedule le pod sur un node
       │
       ▼
5. La tâche s'exécute
       │
       ▼
6. Pod terminé → supprimé (si delete_worker_pods=True)

KubernetesPodOperator

Pour exécuter des tâches avec des images Docker personnalisées :

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from datetime import datetime

with DAG(
    dag_id="etl_with_custom_image",
    start_date=datetime(2024, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:
    
    # Tâche avec image Spark
    spark_job = KubernetesPodOperator(
        task_id="run_spark_etl",
        name="spark-etl-job",
        namespace="airflow",
        image="my-registry/spark-etl:1.0",
        cmds=["spark-submit"],
        arguments=[
            "--master", "k8s://https://kubernetes.default.svc",
            "/app/etl_job.py"
        ],
        # Ressources
        container_resources={
            "requests": {"cpu": "1", "memory": "2Gi"},
            "limits": {"cpu": "2", "memory": "4Gi"},
        },
        # Variables d'environnement
        env_vars={
            "SOURCE_PATH": "s3://bucket/raw/",
            "DEST_PATH": "s3://bucket/processed/",
        },
        # Secrets
        secrets=[
            {"secret": "aws-credentials", "key": "AWS_ACCESS_KEY_ID", "env": "AWS_ACCESS_KEY_ID"},
            {"secret": "aws-credentials", "key": "AWS_SECRET_ACCESS_KEY", "env": "AWS_SECRET_ACCESS_KEY"},
        ],
        # Configuration K8s
        is_delete_operator_pod=True,
        get_logs=True,
        startup_timeout_seconds=300,
        # Affinity/Tolerations
        affinity={
            "nodeAffinity": {
                "requiredDuringSchedulingIgnoredDuringExecution": {
                    "nodeSelectorTerms": [{
                        "matchExpressions": [{
                            "key": "node-type",
                            "operator": "In",
                            "values": ["compute"]
                        }]
                    }]
                }
            }
        },
    )
    
    # Tâche avec image Python custom
    python_job = KubernetesPodOperator(
        task_id="run_python_transform",
        name="python-transform",
        namespace="airflow",
        image="my-registry/python-etl:2.0",
        cmds=["python", "/app/transform.py"],
        container_resources={
            "requests": {"cpu": "500m", "memory": "1Gi"},
            "limits": {"cpu": "1", "memory": "2Gi"},
        },
        is_delete_operator_pod=True,
        get_logs=True,
    )
    
    spark_job >> python_job

pod_template_file : Personnalisation avancée

Pour des configurations complexes, utiliser un template YAML :

# pod_template.yaml
apiVersion: v1
kind: Pod
metadata:
  name: airflow-worker
  labels:
    app: airflow-worker
spec:
  serviceAccountName: airflow-worker
  containers:
    - name: base
      image: apache/airflow:2.8.0-python3.11
      imagePullPolicy: IfNotPresent
      env:
        - name: AIRFLOW__CORE__EXECUTOR
          value: "LocalExecutor"
      resources:
        requests:
          cpu: "500m"
          memory: "512Mi"
        limits:
          cpu: "1000m"
          memory: "1Gi"
      volumeMounts:
        - name: dags
          mountPath: /opt/airflow/dags
          readOnly: true
  volumes:
    - name: dags
      persistentVolumeClaim:
        claimName: airflow-dags
  restartPolicy: Never
  securityContext:
    runAsUser: 50000
    fsGroup: 50000

# Dans le DAG, référencer le template
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

task = KubernetesPodOperator(
    task_id="task_with_template",
    name="custom-task",
    namespace="airflow",
    pod_template_file="/opt/airflow/pod_templates/pod_template.yaml",
    # Override l'image du template
    image="my-custom-image:1.0",
)

2. TaskFlow API

La TaskFlow API (introduite dans Airflow 2.0) permet d'écrire des DAGs de manière Pythonic avec des décorateurs @dag et @task, en gérant automatiquement les XComs.

Pourquoi TaskFlow ?

Exemple comparatif

Avant (approche traditionnelle) :

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def extract(**context):
    data = {"users": 100, "orders": 500}
    context['ti'].xcom_push(key='extracted_data', value=data)
    return data

def transform(**context):
    ti = context['ti']
    data = ti.xcom_pull(task_ids='extract_task', key='extracted_data')
    transformed = {"total": data['users'] + data['orders']}
    ti.xcom_push(key='transformed_data', value=transformed)
    return transformed

def load(**context):
    ti = context['ti']
    data = ti.xcom_pull(task_ids='transform_task', key='transformed_data')
    print(f"Loading: {data}")

with DAG(
    dag_id='traditional_etl',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
) as dag:
    
    extract_task = PythonOperator(
        task_id='extract_task',
        python_callable=extract,
    )
    
    transform_task = PythonOperator(
        task_id='transform_task',
        python_callable=transform,
    )
    
    load_task = PythonOperator(
        task_id='load_task',
        python_callable=load,
    )
    
    extract_task >> transform_task >> load_task

Après (TaskFlow API) :

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

@dag(
    dag_id='taskflow_etl',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
)
def taskflow_etl():
    
    @task
    def extract() -> dict:
        return {"users": 100, "orders": 500}
    
    @task
    def transform(data: dict) -> dict:
        return {"total": data['users'] + data['orders']}
    
    @task
    def load(data: dict):
        print(f"Loading: {data}")
    
    # Dépendances implicites par appel de fonction !
    raw_data = extract()
    transformed_data = transform(raw_data)
    load(transformed_data)

# Instancier le DAG
taskflow_etl()

Avantages TaskFlow

TaskFlow avec plusieurs outputs

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

@dag(
    dag_id='taskflow_multiple_outputs',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
)
def etl_pipeline():
    
    @task(multiple_outputs=True)
    def extract() -> dict:
        """Retourne plusieurs valeurs nommées"""
        return {
            "users": [{"id": 1, "name": "Alice"}],
            "orders": [{"id": 101, "amount": 99.99}],
            "metadata": {"source": "api", "timestamp": "2024-01-01"}
        }
    
    @task
    def process_users(users: list) -> list:
        return [u['name'].upper() for u in users]
    
    @task
    def process_orders(orders: list) -> float:
        return sum(o['amount'] for o in orders)
    
    @task
    def combine(users: list, total: float, metadata: dict):
        print(f"Users: {users}")
        print(f"Total orders: {total}")
        print(f"Source: {metadata['source']}")
    
    # Extraire les données
    data = extract()
    
    # Accéder aux outputs individuels
    processed_users = process_users(data['users'])
    orders_total = process_orders(data['orders'])
    
    # Combiner
    combine(processed_users, orders_total, data['metadata'])

etl_pipeline()

Mixing TaskFlow avec des Operators classiques

from airflow.decorators import dag, task
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from airflow.operators.bash import BashOperator
from datetime import datetime

@dag(
    dag_id='mixed_taskflow',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
)
def mixed_pipeline():
    
    # TaskFlow task
    @task
    def prepare_config() -> dict:
        return {"batch_size": 1000, "date": "2024-01-01"}
    
    # Operator classique
    run_spark = KubernetesPodOperator(
        task_id="run_spark",
        name="spark-job",
        namespace="airflow",
        image="apache/spark:3.5.0",
        cmds=["spark-submit", "/app/job.py"],
    )
    
    # TaskFlow task qui dépend d'un Operator
    @task
    def validate_output():
        print("Validating Spark output...")
        return True
    
    # Bash operator
    notify = BashOperator(
        task_id="notify",
        bash_command="echo 'Pipeline completed!'",
    )
    
    # Définir les dépendances
    config = prepare_config()
    config >> run_spark >> validate_output() >> notify

mixed_pipeline()

3. Dynamic Task Mapping

Le Dynamic Task Mapping (Airflow 2.3+) permet de créer un nombre variable de tâches à runtime, basé sur les données.

Pourquoi Dynamic Mapping ?

Schéma Dynamic Mapping

                    ┌─────────────────┐
                    │  list_files()   │
                    │ [f1, f2, f3, f4]│
                    └────────┬────────┘
                             │
            ┌────────────────┼────────────────┐
            │                │                │
            ▼                ▼                ▼
    ┌───────────────┐ ┌───────────────┐ ┌───────────────┐
    │ process(f1)   │ │ process(f2)   │ │ process(f3)   │ ...
    │   [mapped]    │ │   [mapped]    │ │   [mapped]    │
    └───────┬───────┘ └───────┬───────┘ └───────┬───────┘
            │                 │                 │
            └─────────────────┼─────────────────┘
                              │
                              ▼
                    ┌─────────────────┐
                    │   aggregate()   │
                    │ Combine results │
                    └─────────────────┘

Exemple : Traiter des fichiers dynamiquement

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

@dag(
    dag_id='dynamic_file_processing',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
)
def process_files_dynamically():
    
    @task
    def list_files() -> list[str]:
        """Liste les fichiers à traiter (nombre variable)"""
        # En réalité : lister depuis S3, GCS, etc.
        return [
            "s3://bucket/data/file1.parquet",
            "s3://bucket/data/file2.parquet",
            "s3://bucket/data/file3.parquet",
            "s3://bucket/data/file4.parquet",
        ]
    
    @task
    def process_file(file_path: str) -> dict:
        """Traite UN fichier — sera mappé dynamiquement"""
        print(f"Processing: {file_path}")
        # Simuler le traitement
        row_count = len(file_path) * 100  # Fake
        return {"file": file_path, "rows": row_count}
    
    @task
    def aggregate_results(results: list[dict]) -> dict:
        """Agrège tous les résultats"""
        total_rows = sum(r['rows'] for r in results)
        return {
            "files_processed": len(results),
            "total_rows": total_rows,
        }
    
    # Récupérer la liste de fichiers
    files = list_files()
    
    # 🎯 DYNAMIC MAPPING : .expand() crée N tâches
    processed = process_file.expand(file_path=files)
    
    # Agréger les résultats
    aggregate_results(processed)

process_files_dynamically()

expand() avec plusieurs paramètres

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

@dag(
    dag_id='dynamic_multi_param',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
)
def multi_param_mapping():
    
    @task
    def get_partitions() -> list[dict]:
        return [
            {"date": "2024-01-01", "region": "EU"},
            {"date": "2024-01-01", "region": "US"},
            {"date": "2024-01-02", "region": "EU"},
            {"date": "2024-01-02", "region": "US"},
        ]
    
    @task
    def process_partition(date: str, region: str) -> dict:
        print(f"Processing {region} for {date}")
        return {"date": date, "region": region, "status": "done"}
    
    partitions = get_partitions()
    
    # expand_kwargs pour mapper plusieurs paramètres
    process_partition.expand_kwargs(partitions)

multi_param_mapping()

Limiter le parallélisme

@task(max_active_tis_per_dag=5)  # Max 5 instances en parallèle
def process_file(file_path: str) -> dict:
    # ...
    pass

Mapping sur un Operator (non-TaskFlow)

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id='dynamic_bash',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
) as dag:
    
    # Liste statique (ou XCom d'une tâche précédente)
    files = ["file1.csv", "file2.csv", "file3.csv"]
    
    process = BashOperator.partial(
        task_id="process_files",
    ).expand(
        bash_command=[f"python process.py {f}" for f in files]
    )

4. Airflow vs Dagster vs Prefect

Il existe plusieurs orchestrateurs modernes. Voici une comparaison approfondie.

Vue d'ensemble

Comparaison détaillée

Exemples de code comparés

Même pipeline dans les 3 outils :

Airflow (TaskFlow)

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

@dag(
    dag_id='etl_pipeline',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
)
def etl_pipeline():
    
    @task
    def extract() -> dict:
        return {"data": [1, 2, 3, 4, 5]}
    
    @task
    def transform(raw: dict) -> dict:
        return {"data": [x * 2 for x in raw['data']]}
    
    @task
    def load(transformed: dict):
        print(f"Loading: {transformed}")
    
    raw = extract()
    transformed = transform(raw)
    load(transformed)

etl_pipeline()

Dagster (Asset-based)

from dagster import asset, Definitions, define_asset_job

@asset
def raw_data() -> dict:
    """Asset: données brutes"""
    return {"data": [1, 2, 3, 4, 5]}

@asset
def transformed_data(raw_data: dict) -> dict:
    """Asset: données transformées (dépend de raw_data)"""
    return {"data": [x * 2 for x in raw_data['data']]}

@asset
def loaded_data(transformed_data: dict) -> None:
    """Asset: données chargées"""
    print(f"Loading: {transformed_data}")

# Job pour exécuter tous les assets
etl_job = define_asset_job("etl_job", selection="*")

# Définitions Dagster
defs = Definitions(
    assets=[raw_data, transformed_data, loaded_data],
    jobs=[etl_job],
)

Dagster : Concepts clés

Prefect (Flow-based)

from prefect import flow, task

@task
def extract() -> dict:
    return {"data": [1, 2, 3, 4, 5]}

@task
def transform(raw: dict) -> dict:
    return {"data": [x * 2 for x in raw['data']]}

@task
def load(transformed: dict):
    print(f"Loading: {transformed}")

@flow(name="ETL Pipeline")
def etl_pipeline():
    raw = extract()
    transformed = transform(raw)
    load(transformed)

# Exécuter
if __name__ == "__main__":
    etl_pipeline()

Prefect : Concepts clés

Quand utiliser quoi ?

Tableau de décision

                            Complexité du pipeline
                     Faible ◄─────────────────► Élevée
                        │                          │
    Taille équipe       │                          │
         │              │                          │
     Petite  ───────────┼──► Prefect    ──────────►│ Dagster
         │              │                          │
         │              │                          │
    Grande  ────────────┼──► Prefect    ──────────►│ Airflow
         │              │    ou Dagster            │
         ▼              │                          │

5. OpenLineage — Data Lineage

OpenLineage est un standard ouvert pour collecter et partager les métadonnées de lineage des pipelines data. Il répond à la question : "D'où viennent mes données et où vont-elles ?"

Pourquoi le Data Lineage ?

Schéma OpenLineage

┌─────────────────────────────────────────────────────────────────────────────┐
│                         OPENLINEAGE ECOSYSTEM                               │
│                                                                             │
│   ┌─────────────┐   ┌─────────────┐   ┌─────────────┐   ┌─────────────┐    │
│   │   Airflow   │   │   Spark     │   │    dbt      │   │   Flink     │    │
│   │  + OL Plugin│   │  + OL Lib   │   │  + OL Plugin│   │  + OL Lib   │    │
│   └──────┬──────┘   └──────┬──────┘   └──────┬──────┘   └──────┬──────┘    │
│          │                 │                 │                 │            │
│          │    OpenLineage Events (JSON)      │                 │            │
│          └─────────────────┴─────────────────┴─────────────────┘            │
│                                    │                                        │
│                                    ▼                                        │
│                    ┌───────────────────────────────┐                       │
│                    │     OpenLineage Backend       │                       │
│                    │  (Marquez, Atlan, DataHub...) │                       │
│                    └───────────────────────────────┘                       │
│                                    │                                        │
│                                    ▼                                        │
│                    ┌───────────────────────────────┐                       │
│                    │    Lineage Visualization      │                       │
│                    │    Impact Analysis            │                       │
│                    │    Data Catalog               │                       │
│                    └───────────────────────────────┘                       │
└─────────────────────────────────────────────────────────────────────────────┘

Concepts OpenLineage

Activer OpenLineage dans Airflow

# Installer le provider
pip install apache-airflow-providers-openlineage

# airflow.cfg ou variables d'environnement
[openlineage]
transport = '{"type": "http", "url": "http://marquez:5000", "endpoint": "api/v1/lineage"}'
namespace = "my_airflow_instance"

Ou via environnement :

export AIRFLOW__OPENLINEAGE__TRANSPORT='{"type": "http", "url": "http://marquez:5000", "endpoint": "api/v1/lineage"}'
export AIRFLOW__OPENLINEAGE__NAMESPACE="my_airflow_instance"

Marquez : Backend OpenLineage

Marquez est le backend de référence pour OpenLineage (open-source par WeWork/Linux Foundation).

# Déployer Marquez avec Docker Compose
git clone https://github.com/MarquezProject/marquez.git
cd marquez
docker-compose up -d

# UI disponible sur http://localhost:3000
# API sur http://localhost:5000

DAG avec lineage explicite

from airflow.decorators import dag, task
from airflow.lineage.entities import Table, File
from datetime import datetime

@dag(
    dag_id='etl_with_lineage',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
)
def etl_with_lineage():
    
    @task(
        inlets=[File(url="s3://bucket/raw/")],
        outlets=[Table(cluster="warehouse", database="analytics", name="staging_orders")],
    )
    def extract_and_stage():
        """Extract depuis S3 et charge dans staging"""
        print("Extracting from S3 to staging...")
    
    @task(
        inlets=[Table(cluster="warehouse", database="analytics", name="staging_orders")],
        outlets=[Table(cluster="warehouse", database="analytics", name="fact_orders")],
    )
    def transform_to_fact():
        """Transforme staging en fact table"""
        print("Transforming to fact table...")
    
    extract_and_stage() >> transform_to_fact()

etl_with_lineage()

Lineage automatique avec certains Operators

Certains operators extraient automatiquement le lineage :

Alternatives à Marquez

6. Astronomer — Airflow Enterprise

Astronomer est la plateforme enterprise pour Apache Airflow. Elle fournit des outils, un hosting managé, et des SDKs pour industrialiser Airflow.

Pourquoi Astronomer ?

Écosystème Astronomer

┌─────────────────────────────────────────────────────────────────────────────┐
│                         ASTRONOMER ECOSYSTEM                                │
│                                                                             │
│   ┌─────────────────────────────────────────────────────────────────────┐  │
│   │                        Astro Cloud                                   │  │
│   │  • Airflow fully managed                                            │  │
│   │  • Auto-scaling                                                      │  │
│   │  • Observability intégrée                                           │  │
│   │  • CI/CD intégré                                                    │  │
│   └─────────────────────────────────────────────────────────────────────┘  │
│                                                                             │
│   ┌───────────────────────┐   ┌───────────────────────┐                    │
│   │      Astro CLI        │   │      Astro SDK        │                    │
│   │  • Dev local          │   │  • API Pythonic       │                    │
│   │  • Deploy en 1 cmd    │   │  • SQL/Python tasks   │                    │
│   │  • Tests intégrés     │   │  • Data quality       │                    │
│   └───────────────────────┘   └───────────────────────┘                    │
│                                                                             │
│   ┌─────────────────────────────────────────────────────────────────────┐  │
│   │                    Astronomer Registry                               │  │
│   │  • Providers certifiés                                               │  │
│   │  • DAGs de référence                                                 │  │
│   │  • Documentation enrichie                                            │  │
│   └─────────────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────────┘

Astro CLI

L'outil en ligne de commande pour développer et déployer Airflow.

# Installation
# macOS
brew install astro

# Linux
curl -sSL install.astronomer.io | sudo bash -s

# Windows (WSL)
curl -sSL install.astronomer.io | sudo bash -s

# Créer un nouveau projet Airflow
mkdir my-airflow-project && cd my-airflow-project
astro dev init

# Structure générée :
# .
# ├── dags/                 # Tes DAGs
# │   └── example_dag.py
# ├── include/              # Fichiers additionnels
# ├── plugins/              # Plugins custom
# ├── tests/                # Tests
# ├── Dockerfile            # Image Airflow custom
# ├── packages.txt          # Packages système
# ├── requirements.txt      # Dépendances Python
# └── airflow_settings.yaml # Variables, connexions

# Démarrer l'environnement local
astro dev start
# UI sur http://localhost:8080 (admin/admin)

# Voir les logs
astro dev logs

# Exécuter les tests
astro dev pytest

# Parser les DAGs (vérifier les erreurs)
astro dev parse

# Arrêter
astro dev stop

# Déployer sur Astro Cloud
astro deploy

Astro SDK

Une API Python simplifiée pour écrire des DAGs, particulièrement pour les opérations SQL/dataframe.

pip install astro-sdk-python

Exemple : ETL SQL simplifié

from datetime import datetime
from airflow.decorators import dag
from astro import sql as aql
from astro.files import File
from astro.sql.table import Table

@dag(
    dag_id='astro_sdk_etl',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
)
def astro_sdk_etl():
    
    # Charger un CSV dans une table
    raw_orders = aql.load_file(
        input_file=File("s3://bucket/orders.csv"),
        output_table=Table(
            name="raw_orders",
            conn_id="postgres_conn",
        ),
    )
    
    # Transformation SQL (inline)
    @aql.transform
    def transform_orders(input_table: Table):
        return """
            SELECT 
                order_id,
                customer_id,
                order_date,
                amount,
                amount * 1.2 as amount_with_tax
            FROM {{ input_table }}
            WHERE amount > 0
        """
    
    # Appliquer la transformation
    transformed = transform_orders(
        input_table=raw_orders,
        output_table=Table(
            name="transformed_orders",
            conn_id="postgres_conn",
        ),
    )
    
    # Exporter vers S3
    aql.export_to_file(
        input_data=transformed,
        output_file=File("s3://bucket/processed/orders.parquet"),
        if_exists="replace",
    )
    
    # Nettoyer les tables temporaires
    aql.cleanup()

astro_sdk_etl()

Fonctionnalités Astro SDK

Astro SDK : Transformation DataFrame

from astro import sql as aql
from astro.sql.table import Table
import pandas as pd

@aql.dataframe
def process_with_pandas(df: pd.DataFrame) -> pd.DataFrame:
    """Transformation avec Pandas"""
    df['total'] = df['quantity'] * df['price']
    df['processed_at'] = pd.Timestamp.now()
    return df

# Dans le DAG :
processed = process_with_pandas(
    df=some_table,
    output_table=Table(name="processed", conn_id="postgres"),
)

Astro Cloud vs Self-Hosted

7. Exercices Pratiques

Exercice 1 : DAG TaskFlow avec Dynamic Mapping

Objectif : Créer un DAG qui traite dynamiquement une liste de fichiers.

Instructions :

1. Créer une tâche list_files() qui retourne une liste de chemins
2. Créer une tâche process_file(path) mappée dynamiquement
3. Créer une tâche report(results) qui agrège les résultats
4. Limiter à 3 tâches parallèles maximum

Exercice 2 : KubernetesPodOperator avec Spark

Objectif : Créer un DAG qui exécute un job Spark via KubernetesPodOperator.

Instructions :

1. Utiliser l'image apache/spark:3.5.0
2. Configurer les ressources : 2 CPU, 4Gi RAM
3. Passer des variables d'environnement pour la config
4. Utiliser une node affinity pour cibler les nodes compute

Exercice 3 : Comparatif Airflow/Dagster/Prefect

Objectif : Implémenter le même pipeline simple dans les 3 outils.

Pipeline :

1. Lire un fichier JSON
2. Filtrer les enregistrements (amount > 100)
3. Calculer une somme
4. Écrire le résultat

Comparer :

• Nombre de lignes de code
• Facilité de test local
• Lisibilité

Exercice 4 : OpenLineage avec Marquez

Objectif : Configurer OpenLineage et visualiser le lineage.

Instructions :

1. Déployer Marquez en local (Docker Compose)
2. Configurer Airflow pour envoyer les events à Marquez
3. Créer un DAG avec inlets et outlets explicites
4. Visualiser le lineage dans l'UI Marquez

Exercice 5 : Astro CLI Project

Objectif : Créer un projet Airflow complet avec Astro CLI.

Instructions :

1. Initialiser un projet avec astro dev init
2. Créer un DAG utilisant Astro SDK
3. Ajouter des tests dans /tests
4. Valider avec astro dev pytest

8. Mini-Projet : Pipeline Data Production-Ready

Objectif

Créer un pipeline ETL production-ready déployé sur Kubernetes avec monitoring et lineage.

Architecture cible

┌─────────────────────────────────────────────────────────────────────────────┐
│                          MINI-PROJET M28                                    │
│                                                                             │
│   ┌─────────────────────────────────────────────────────────────────────┐  │
│   │                    Airflow sur Kubernetes                            │  │
│   │                                                                      │  │
│   │   ┌─────────────────────────────────────────────────────────────┐   │  │
│   │   │            DAG: data_pipeline_advanced                       │   │  │
│   │   │                                                              │   │  │
│   │   │   ┌─────────┐    ┌─────────────┐    ┌─────────────────┐     │   │  │
│   │   │   │ Extract │───▶│  Transform  │───▶│      Load       │     │   │  │
│   │   │   │ (K8sPod)│    │ (TaskFlow)  │    │  (Astro SDK)    │     │   │  │
│   │   │   └─────────┘    └──────┬──────┘    └─────────────────┘     │   │  │
│   │   │                         │                                    │   │  │
│   │   │            Dynamic Mapping (N fichiers)                      │   │  │
│   │   └─────────────────────────────────────────────────────────────┘   │  │
│   │                                                                      │  │
│   │   ┌─────────────┐   ┌─────────────┐   ┌─────────────────────────┐   │  │
│   │   │  Scheduler  │   │  Webserver  │   │    Worker Pods (K8s)    │   │  │
│   │   └─────────────┘   └─────────────┘   └─────────────────────────┘   │  │
│   └─────────────────────────────────────────────────────────────────────┘  │
│                                    │                                        │
│                         OpenLineage Events                                  │
│                                    ▼                                        │
│   ┌─────────────────────────────────────────────────────────────────────┐  │
│   │                         Marquez                                      │  │
│   │              (Data Lineage Visualization)                           │  │
│   └─────────────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────────┘

Instructions

Étape 1 : Setup Astro Project

mkdir advanced-pipeline && cd advanced-pipeline
astro dev init

Étape 2 : Créer le DAG principal

Le DAG doit :

• Lister des fichiers depuis S3 (simulé)
• Traiter chaque fichier avec Dynamic Mapping
• Utiliser TaskFlow API
• Avoir des inlets/outlets pour le lineage

Étape 3 : Configurer OpenLineage

• Ajouter la config dans airflow_settings.yaml
• Vérifier que les events arrivent dans Marquez

Étape 4 : Tests

• Écrire des tests unitaires pour les tâches
• Valider avec astro dev pytest

Étape 5 : Déployer (optionnel)

• Déployer sur un cluster K8s avec le Helm chart
• Ou utiliser Astro Cloud

✅ Solution du mini-projet

advanced-pipeline/
├── dags/
│   └── data_pipeline_advanced.py
├── include/
│   └── sql/
│       └── transform.sql
├── tests/
│   └── test_pipeline.py
├── Dockerfile
├── requirements.txt
└── airflow_settings.yaml

astro-sdk-python>=1.5.0
apache-airflow-providers-openlineage>=1.0.0
apache-airflow-providers-cncf-kubernetes>=7.0.0
pandas>=2.0.0

airflow:
  connections:
    - conn_id: postgres_conn
      conn_type: postgres
      host: postgres
      login: airflow
      password: airflow
      schema: airflow
      port: 5432

  variables:
    - variable_name: data_bucket
      variable_value: s3://my-data-bucket

  # OpenLineage config
  openlineage:
    transport: '{"type": "http", "url": "http://marquez:5000", "endpoint": "api/v1/lineage"}'
    namespace: advanced-pipeline

from datetime import datetime
from airflow.decorators import dag, task
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from airflow.lineage.entities import File, Table
from astro import sql as aql
from astro.sql.table import Table as AstroTable
import pandas as pd

@dag(
    dag_id='data_pipeline_advanced',
    start_date=datetime(2024, 1, 1),
    schedule='@daily',
    catchup=False,
    tags=['production', 'etl'],
    doc_md="""## Pipeline Data Avancé
    
    Ce pipeline:
    1. Liste les fichiers à traiter
    2. Extrait chaque fichier (Dynamic Mapping)
    3. Transforme avec Astro SDK
    4. Charge dans PostgreSQL
    """,
)
def data_pipeline_advanced():
    
    # --- ÉTAPE 1 : Lister les fichiers ---
    @task
    def list_files() -> list[str]:
        """Liste les fichiers à traiter (simulé)"""
        # En production: lister depuis S3
        return [
            "orders_2024_01.parquet",
            "orders_2024_02.parquet",
            "orders_2024_03.parquet",
        ]
    
    # --- ÉTAPE 2 : Extraire chaque fichier ---
    @task(
        max_active_tis_per_dag=3,
        inlets=[File(url="s3://bucket/raw/")],
    )
    def extract_file(file_name: str) -> dict:
        """Extrait un fichier (simulé)"""
        print(f"Extracting: {file_name}")
        # Simuler des données
        return {
            "file": file_name,
            "rows": 1000,
            "data": [
                {"order_id": 1, "amount": 100},
                {"order_id": 2, "amount": 200},
            ]
        }
    
    # --- ÉTAPE 3 : Transformer avec Pandas ---
    @task(
        outlets=[Table(cluster="warehouse", database="analytics", name="orders_processed")],
    )
    def transform_data(extracted_list: list[dict]) -> list[dict]:
        """Agrège et transforme les données"""
        all_data = []
        for extracted in extracted_list:
            for record in extracted['data']:
                record['source_file'] = extracted['file']
                record['amount_with_tax'] = record['amount'] * 1.2
                all_data.append(record)
        
        print(f"Transformed {len(all_data)} records")
        return all_data
    
    # --- ÉTAPE 4 : Charger dans PostgreSQL ---
    @aql.dataframe
    def load_to_postgres(data: list[dict]) -> pd.DataFrame:
        """Charge dans PostgreSQL via Astro SDK"""
        df = pd.DataFrame(data)
        df['loaded_at'] = pd.Timestamp.now()
        return df
    
    # --- ÉTAPE 5 : Rapport final ---
    @task
    def generate_report(row_count: int):
        """Génère un rapport"""
        print(f"Pipeline completed. Total rows: {row_count}")
    
    # --- ORCHESTRATION ---
    files = list_files()
    
    # Dynamic mapping sur les fichiers
    extracted = extract_file.expand(file_name=files)
    
    # Transformer
    transformed = transform_data(extracted)
    
    # Charger
    loaded = load_to_postgres(
        data=transformed,
        output_table=AstroTable(
            name="orders_processed",
            conn_id="postgres_conn",
        ),
    )
    
    # Rapport
    generate_report(len(transformed))
    
    # Cleanup Astro SDK
    aql.cleanup()

# Instancier le DAG
data_pipeline_advanced()

import pytest
from dags.data_pipeline_advanced import data_pipeline_advanced

def test_dag_loads():
    """Test que le DAG se charge sans erreur"""
    dag = data_pipeline_advanced()
    assert dag is not None
    assert dag.dag_id == "data_pipeline_advanced"

def test_dag_has_expected_tasks():
    """Test que le DAG a les bonnes tâches"""
    dag = data_pipeline_advanced()
    task_ids = [t.task_id for t in dag.tasks]
    
    assert "list_files" in task_ids
    assert "transform_data" in task_ids

def test_extract_file():
    """Test unitaire de la fonction extract"""
    # Import direct de la fonction
    from dags.data_pipeline_advanced import extract_file
    
    # Appeler la fonction wrapped
    result = extract_file.function("test.parquet")
    
    assert "file" in result
    assert "rows" in result
    assert result["file"] == "test.parquet"

# Démarrer l'environnement
astro dev start

# Parser les DAGs
astro dev parse

# Lancer les tests
astro dev pytest

# Voir l'UI
open http://localhost:8080

# Déclencher le DAG
astro dev run dags trigger data_pipeline_advanced

version: "3"
services:
  marquez:
    image: marquezproject/marquez:latest
    ports:
      - "5000:5000"
      - "5001:5001"
    environment:
      - MARQUEZ_PORT=5000
      - MARQUEZ_ADMIN_PORT=5001
  
  marquez-web:
    image: marquezproject/marquez-web:latest
    ports:
      - "3000:3000"
    environment:
      - MARQUEZ_HOST=marquez
      - MARQUEZ_PORT=5000

📚 Ressources pour aller plus loin

🌐 Documentation officielle

• Apache Airflow Docs — Documentation Airflow
• Airflow Helm Chart — Déploiement K8s
• Dagster Docs — Documentation Dagster
• Prefect Docs — Documentation Prefect
• OpenLineage — Standard de lineage
• Astronomer Docs — Documentation Astronomer

🎮 Pratique

• Astronomer Academy — Cours gratuits Airflow
• Astro CLI Quickstart — Démarrer avec Astro
• Marquez Demo — Essayer OpenLineage

📖 Livres & Articles

• Data Pipelines with Apache Airflow — Bas Harenslak & Julian de Ruiter
• Fundamentals of Data Engineering — Joe Reis & Matt Housley
• Astronomer Blog — Best practices Airflow

🔧 Outils

• Astronomer Registry — Providers et DAGs
• Marquez — Backend OpenLineage
• DataHub — Data catalog avec lineage

➡️ Prochaine étape

Maintenant que tu maîtrises l'orchestration avancée, passons au messaging distribué !

👉 Module suivant : 29_distributed_messaging — Kafka avancé, RabbitMQ, Pulsar, Debezium

Tu vas apprendre :

• Kafka avancé (Quotas, Tiered Storage)
• Alternatives : RabbitMQ, Apache Pulsar
• Change Data Capture avec Debezium
• Patterns de messaging distribué

🎉 Félicitations ! Tu as terminé le module Advanced Orchestration pour Data Engineers.