Dans cet atelier applicatif, nous allons concevoir et déployer trois pipelines de données d'entreprise prêts pour la production, illustrant les modèles de conception (Design Patterns) les plus demandés en ingénierie de données.
flowchart TD
subgraph App1["Application 1 : Ingestion Datamart"]
A1["Extraction Base Transactionnelle"] --> A2["Transformation & Calculs Pandas"] --> A3["Chargement par Lots dans Postgres"]
end
subgraph App2["Application 2 : Aiguillage Qualité"]
B1["Ingestion Données"] --> B2{"Contrôle Qualité<br><i>(@task.branch)</i>"}
B2 -->|Valide| B3["Promotion en Production"]
B2 -->|Anomalie| B4["Quarantaine & Alerte"]
end
subgraph App3["Application 3 : Tâches Dynamiques"]
C1["Cartographie Dynamique<br><i>Dynamic Task Mapping (.expand)</i>"] --> C2["Exécution Parallèle Multi-Régions"]
end
style App1 fill:#e0f2fe,stroke:#0284c7,stroke-width:2px,color:#0f172a
style App2 fill:#fef3c7,stroke:#d97706,stroke-width:2px,color:#0f172a
style App3 fill:#dcfce7,stroke:#16a34a,stroke-width:2px,color:#0f172a
Application 1 : Pipeline ETL d'actualisation quotidienne de Datamart#
Dans le secteur des télécommunications et du retail, les équipes marketing ont besoin de consulter chaque matin les métriques agrégées de consommation et de recharges de la veille.
Ce pipeline extrait les transactions brutes de la base de production (OLTP), nettoie et agrège les indicateurs en mémoire avec Pandas, puis réalise une insertion performante par lots (chunksize) dans l'entrepôt décisionnel.
dags/app_01_etl_datamart.py
"""
Pipeline ETL Datamart : Consolidation quotidienne des ventes et recharges.
Auteur : Joseph Konka (AlgoJungle)
"""
from datetime import datetime, timedelta
import pandas as pd
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
# Configuration et constantes
SRC_CONNECTION_ID = "postgres_oltp"
DEST_CONNECTION_ID = "postgres_default"
DEST_TABLE = "datamart.kpi_ventes_journalieres"
default_args = {
"owner": "data_ops",
"start_date": datetime(2026, 1, 1),
"retries": 2,
"retry_delay": timedelta(minutes=5),
}
def extraire_et_charger_datamart(**context):
"""Extrait les ventes de la veille et actualise la table cible de manière idempotente."""
# Récupération de la date logique d'exécution (format YYYY-MM-DD)
date_traitement = context["ds"]
print(f"Lancement de la consolidation pour la date : {date_traitement}")
# 1. Extraction depuis la source OLTP
hook_source = PostgresHook(postgres_conn_id=SRC_CONNECTION_ID)
requete_extraction = f"""
SELECT
client_id,
categorie_produit,
SUM(montant) AS total_montant,
COUNT(transaction_id) AS nb_transactions,
'{date_traitement}'::DATE AS date_consolidation
FROM raw_transactions
WHERE date_transaction::DATE = '{date_traitement}'::DATE
GROUP BY client_id, categorie_produit;
"""
df = hook_source.get_pandas_df(requete_extraction)
print(f"Extraction terminée : {len(df)} lignes agrégées.")
if df.empty:
print("Aucune transaction enregistrée pour cette date. Fin du traitement.")
return
# 2. Nettoyage et enrichissement Pandas
df["total_montant"] = df["total_montant"].fillna(0.0)
df["panier_moyen"] = df["total_montant"] / df["nb_transactions"]
df["date_maj"] = datetime.now()
# 3. Chargement idempotent dans la base cible
hook_dest = PostgresHook(postgres_conn_id=DEST_CONNECTION_ID)
# Suppression préalable des données de cette date pour garantir l'idempotence
suppression_sql = f"DELETE FROM {DEST_TABLE} WHERE date_consolidation = '{date_traitement}'::DATE;"
hook_dest.run(suppression_sql)
# Insertion en masse par paquets de 10 000 lignes
schema_name, table_name = DEST_TABLE.split(".")
df.to_sql(
name=table_name,
schema=schema_name,
con=hook_dest.get_sqlalchemy_engine(),
if_exists="append",
index=False,
chunksize=10000,
method="multi",
)
print(f"Chargement réussi de {len(df)} enregistrements dans {DEST_TABLE} !")
with DAG(
dag_id="app_01_etl_datamart",
description="Consolidation quotidienne des recharges dans le Datamart",
default_args=default_args,
schedule="30 5 * * *", # Tous les matins à 05h30 UTC
catchup=False,
max_active_runs=1,
tags=["production", "telecom", "etl", "datamart"],
) as dag:
debut = EmptyOperator(task_id="debut")
# Création du schéma et de la table si nécessaire
preparer_table = SQLExecuteQueryOperator(
task_id="preparer_table_cible",
conn_id=DEST_CONNECTION_ID,
sql=f"""
CREATE SCHEMA IF NOT EXISTS datamart;
CREATE TABLE IF NOT EXISTS {DEST_TABLE} (
client_id VARCHAR(50),
categorie_produit VARCHAR(100),
total_montant NUMERIC(14,2),
nb_transactions INTEGER,
date_consolidation DATE,
panier_moyen NUMERIC(10,2),
date_maj TIMESTAMP
);
""",
)
tache_etl = PythonOperator(
task_id="extraire_transformer_charger",
python_callable=extraire_et_charger_datamart,
)
fin = EmptyOperator(task_id="fin")
debut >> preparer_table >> tache_etl >> fin
[!TIP] Le secret de l'idempotence : Notez l'instruction
DELETE FROM ... WHERE date_consolidation = ...exécutée avant l'insertion. Si le pipeline est relancé 3 fois pour la même date, la table ne contiendra jamais de doublons. Votre pipeline est idempotent, ce qui est le premier critère de qualité en Data Engineering.
Application 2 : Contrôle qualité et aiguillage conditionnel (TaskFlow API)#
Dans un environnement de données rigoureux, des données corrompues ne doivent jamais atteindre la production.
Ce pipeline télécharge des fichiers de commandes, calcule un taux d'anomalie (valeurs nulles, montants négatifs), et prend une décision via @task.branch : si le fichier est propre, il est promu en production ; si des anomalies dépassent le seuil toléré, il est dérouté vers une table de quarantaine et une alerte est émise.
dags/app_02_controle_qualite_branching.py
"""
Pipeline avec contrôle qualité et aiguillage conditionnel.
Auteur : Joseph Konka (AlgoJungle)
"""
from datetime import datetime
import pandas as pd
from airflow.decorators import dag, task
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule
SEUIL_ANOMALIE_MAX = 0.05 # Maximum 5 % d'erreurs autorisées
@dag(
dag_id="app_02_controle_qualite_branching",
start_date=datetime(2026, 1, 1),
schedule="@daily",
catchup=False,
tags=["qualite", "branching", "taskflow"],
)
def pipeline_qualite_donnees():
@task
def ingerer_donnees_entrantes() -> dict:
"""Simule l'ingestion d'un lot de données."""
donnees = [
{"id": 1, "client": "Dupont", "montant": 150.0},
{"id": 2, "client": "Martin", "montant": -20.0}, # Anomalie : montant négatif
{"id": 3, "client": None, "montant": 45.0}, # Anomalie : client null
{"id": 4, "client": "Durand", "montant": 210.0},
{"id": 5, "client": "Lefebvre", "montant": 80.0},
]
return {"lignes": donnees}
@task.branch
def auditer_qualite(payload: dict) -> str:
"""Vérifie l'intégrité et choisit la branche d'exécution."""
df = pd.DataFrame(payload["lignes"])
nb_total = len(df)
# Détection des anomalies
anomalies = df[df["montant"] <= 0 | df["client"].isna()]
taux_erreur = len(anomalies) / nb_total
print(f"Rapport Qualité : {len(anomalies)} anomalies détectées ({taux_erreur:.1%}).")
if taux_erreur > SEUIL_ANOMALIE_MAX:
print("Seuil d'anomalie dépassé ! Aiguillage vers la quarantaine.")
return "isoler_en_quarantaine"
else:
print("Qualité conforme. Aiguillage vers la promotion en production.")
return "publier_en_production"
@task
def publier_en_production():
print("🟢 Données insérées dans les tables analytiques de production.")
@task
def isoler_en_quarantaine():
print("🔴 Données isolées dans la table 'quarantaine_rejets' pour inspection par le Data Steward.")
# Tâche finale avec trigger rule permettant de joindre les branches
cloture = EmptyOperator(
task_id="cloture_pipeline",
trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
)
# Déclaration du flux
donnees = ingerer_donnees_entrantes()
decision = auditer_qualite(donnees)
branche_prod = publier_en_production()
branche_quarantaine = isoler_en_quarantaine()
decision >> [branche_prod, branche_quarantaine] >> cloture
# Instanciation du DAG
pipeline_qualite_donnees()
Application 3 : Traitement parallèle multi-territoires (Dynamic Task Mapping)#
Lorsque votre entreprise est présente dans 20 pays, créer manuellement 20 tâches d'ingestion identiques dans votre fichier .py est une mauvaise pratique.
Airflow propose le Dynamic Task Mapping (.expand()), qui instancie automatiquement à l'exécution une tâche parallèle pour chaque élément d'une liste dynamique :
dags/app_03_dynamic_task_mapping.py
"""
Pipeline avec génération dynamique de tâches (Dynamic Task Mapping).
Auteur : Joseph Konka (AlgoJungle)
"""
from datetime import datetime
from airflow.decorators import dag, task
@dag(
dag_id="app_03_dynamic_task_mapping",
start_date=datetime(2026, 1, 1),
schedule="@weekly",
catchup=False,
tags=["dynamic-mapping", "international", "parallelisme"],
)
def pipeline_multi_pays():
@task
def lister_pays_actifs() -> list[str]:
"""Récupère dynamiquement la liste des filiales actives."""
# En production, cette liste peut être issue d'une requête SQL ou d'une API
return ["France", "Allemagne", "Espagne", "Italie", "Royaume-Uni", "Canada"]
@task
def traiter_filiale(pays: str) -> dict:
"""Cette tâche sera dupliquée dynamiquement pour chaque pays !"""
print(f"Traitement des métriques financières pour le pays : {pays}")
ca_calcule = len(pays) * 100000 # Simulation d'un calcul métier
return {"pays": pays, "chiffre_affaires": ca_calcule}
@task
def consolider_resultats(resultats: list[dict]):
"""Regroupe les résultats de toutes les tâches parallèles générées."""
total_global = sum(r["chiffre_affaires"] for r in resultats)
print(f"Consolidation mondiale terminée ! Chiffre d'affaires total : {total_global:,.2f} €")
# Dynamic Task Mapping : .expand() instancie automatiquement autant de workers que de pays !
liste_pays = lister_pays_actifs()
resultats_pays = traiter_filiale.expand(pays=liste_pays)
consolider_resultats(resultats_pays)
pipeline_multi_pays()
Dans l'interface web d'Airflow, la tâche traiter_filiale s'affichera sous la forme traiter_filiale [6], représentant les 6 instances exécutées en parallèle sur les workers Celery !
Grille d'évaluation des pipelines de production#
Avant de soumettre un DAG en revue de code dans une équipe Data, assurez-vous de respecter ces critères d'excellence :
- Idempotence garantie : Le pipeline produit le même résultat exact, qu'il soit exécuté une fois ou cinq fois pour la même date logique.
- Aucun secret en clair : Toutes les bases de données et clés API passent par des
conn_idou des variables d'environnement chiffrées. - Légèreté du fichier DAG : Le fichier
.pyne réalise aucun calcul lourd au niveau racine (les calculs sont confinés à l'intérieur des fonctions des tâches). - Gestion des pannes :
retriesetretry_delaysont configurés pour absorber les instabilités réseau passagères. - Documentation et Tags : Le DAG comporte une description claire, un propriétaire (
owner) et des tags pour faciliter la navigation.
Félicitations pour la réalisation de ces trois architectures de référence ! Dans la dernière section, nous passons en revue les ressources pour continuer à progresser et un quiz de certification des acquis.
Commentaires