Pour ce premier projet pratique, nous allons construire un pipeline complet d'extraction, de transformation et de chargement (ETL). Notre objectif consiste à ingérer quotidiennement des indicateurs sanitaires publics mondiaux (cas confirmés, guérisons et décès par pays) depuis un référentiel distant, à nettoyer les données et à charger les métriques consolidées dans une table relationnelle PostgreSQL.

---

1. Deux manières d'écrire un DAG dans Airflow

Depuis l'avènement d'Airflow 2.0, vous disposez de deux styles d'écriture :

flowchart TD
    subgraph Classique["Approche 1 : Opérateurs Classiques"]
        C1["PythonOperator(task_id='...', python_callable=...)"]
        C2["SQLExecuteQueryOperator(task_id='...', conn_id=...)"]
        C3["Gestion manuelle des XComs (xcom_push, xcom_pull)"]
    end
    subgraph TaskFlow["Approche 2 : TaskFlow API (Moderne)"]
        T1["Décorateur @dag au-dessus de la fonction de pipeline"]
        T2["Décorateur @task au-dessus des fonctions métiers Python"]
        T3["Transmission naturelle des données par retours de fonctions"]
    end
    style Classique fill:#f1f5f9,stroke:#64748b
    style TaskFlow fill:#dcfce7,stroke:#16a34a,stroke-width:2px,color:#0f172a

Nous allons examiner ces deux approches afin que vous soyez à l'aise sur n'importe quelle base de code en entreprise.

---

2. Implémentation avec les Opérateurs Classiques

Créez le fichier dags/etl_covid19_daily.py dans votre dossier local dags/ :

dags/etl_covid19_daily.py

"""
Pipeline ETL : Ingestion quotidienne des statistiques sanitaires mondiales.
Auteur : Joseph Konka (AlgoJungle)
"""

from datetime import datetime, timedelta
import pandas as pd
import requests

from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook

# 1. Paramètres par défaut partagés par toutes les tâches
default_args = {
    "owner": "algojungle",
    "depends_on_past": False,
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": False,
}

# 2. Fonctions métiers d'extraction, transformation et chargement
def extraire_donnees_brutes(ti) -> str:
    """Télécharge le fichier CSV brut et stocke le chemin local temporaire."""
    url = "https://raw.githubusercontent.com/datasets/covid-19/main/data/countries-aggregated.csv"
    chemin_temporaire = "/opt/airflow/logs/donnees_brutes.csv"

    reponse = requests.get(url, timeout=30)
    reponse.raise_for_status()

    with open(chemin_temporaire, "w", encoding="utf-8") as f:
        f.write(reponse.text)

    print(f"Extraction réussie. Fichier enregistré sous : {chemin_temporaire}")
    # Le retour de fonction est automatiquement transmis via XCom
    return chemin_temporaire

def transformer_donnees(ti) -> str:
    """Nettoie les données et calcule les indicateurs dérivés."""
    # Récupération du chemin du fichier produit par la tâche précédente
    chemin_brut = ti.xcom_pull(task_ids="extraire_donnees")
    chemin_transforme = "/opt/airflow/logs/donnees_propres.csv"

    df = pd.read_csv(chemin_brut)

    # Renommage et filtrage
    df.columns = [col.lower().replace(" ", "_") for col in df.columns]
    df["date"] = pd.to_datetime(df["date"]).dt.date

    # Remplacement des valeurs manquantes
    df["confirmed"] = df["confirmed"].fillna(0).astype(int)
    df["recovered"] = df.get("recovered", 0).fillna(0).astype(int)
    df["deaths"] = df["deaths"].fillna(0).astype(int)

    # Calcul de cas actifs et du ratio de létalité
    df["active"] = df["confirmed"] - (df["recovered"] + df["deaths"])
    df["fatality_rate"] = (df["deaths"] / df["confirmed"].replace(0, 1)) * 100

    # Ne conserver que le dernier instantané disponible par pays
    df_dernier = df.sort_values("date").groupby("country").last().reset_index()
    df_dernier.to_csv(chemin_transforme, index=False)

    print(f"Transformation terminée pour {len(df_dernier)} pays.")
    return chemin_transforme

def charger_dans_postgres(ti):
    """Charge les données transformées dans PostgreSQL via le PostgresHook."""
    chemin_transforme = ti.xcom_pull(task_ids="transformer_donnees")
    df = pd.read_csv(chemin_transforme)

    hook = PostgresHook(postgres_conn_id="postgres_default")
    moteur = hook.get_sqlalchemy_engine()

    # Écriture dans la table PostgreSQL
    df.to_sql(
        name="kpi_covid_mondial",
        con=moteur,
        schema="public",
        if_exists="replace",
        index=False,
    )
    print("Données insérées avec succès dans PostgreSQL !")

# 3. Déclaration du DAG via le bloc contextuel
with DAG(
    dag_id="etl_covid19_daily",
    description="Pipeline ETL quotidien des métriques de santé",
    default_args=default_args,
    start_date=datetime(2026, 1, 1),
    schedule="30 6 * * *",  # Exécution tous les jours à 06h30 UTC
    catchup=False,
    tags=["sante", "covid19", "etl", "production"],
) as dag:

    debut = EmptyOperator(task_id="debut")

    creer_table = SQLExecuteQueryOperator(
        task_id="creer_table",
        conn_id="postgres_default",
        sql="""
        CREATE TABLE IF NOT EXISTS public.kpi_covid_mondial (
            country VARCHAR(150),
            date DATE,
            confirmed INTEGER,
            recovered INTEGER,
            deaths INTEGER,
            active INTEGER,
            fatality_rate NUMERIC,
            charge_le TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
        """,
    )

    tache_extraction = PythonOperator(
        task_id="extraire_donnees",
        python_callable=extraire_donnees_brutes,
    )

    tache_transformation = PythonOperator(
        task_id="transformer_donnees",
        python_callable=transformer_donnees,
    )

    tache_chargement = PythonOperator(
        task_id="charger_donnees",
        python_callable=charger_dans_postgres,
    )

    fin = EmptyOperator(task_id="fin")

    # 4. Orchestration et ordonnancement des dépendances
    debut >> creer_table >> tache_extraction >> tache_transformation >> tache_chargement >> fin

---

3. L'alternative moderne : La TaskFlow API (@dag, @task)

Dans les projets récents, la TaskFlow API simplifie considérablement la syntaxe en transformant de simples fonctions Python en tâches Airflow grâce à des décorateurs. Plus besoin d'écrire manuellement des xcom_pull() ou de déclarer des PythonOperator :

dags/etl_taskflow_moderne.py

from datetime import datetime, timedelta
import pandas as pd
import requests

from airflow.decorators import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook

@dag(
    dag_id="etl_covid19_taskflow",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["taskflow", "moderne", "etl"],
)
def pipeline_covid_taskflow():

    @task
    def extraire() -> dict:
        """Récupère les données et renvoie un payload structuré."""
        url = "https://raw.githubusercontent.com/datasets/covid-19/main/data/countries-aggregated.csv"
        df = pd.read_csv(url)
        return df.tail(100).to_dict(orient="records")

    @task
    def transformer(donnees: dict) -> dict:
        """Nettoie le dictionnaire reçu en argument."""
        df = pd.DataFrame(donnees)
        df["active"] = df["Confirmed"] - df["Deaths"]
        return df.to_dict(orient="records")

    @task
    def charger(donnees_propres: dict):
        """Charge le résultat dans PostgreSQL."""
        df = pd.DataFrame(donnees_propres)
        hook = PostgresHook(postgres_conn_id="postgres_default")
        df.to_sql("kpi_taskflow", con=hook.get_sqlalchemy_engine(), if_exists="replace", index=False)
        print("Chargement TaskFlow réussi !")

    # Les dépendances et le passage de XCom s'effectuent naturellement comme du code Python pur !
    donnees_brutes = extraire()
    donnees_transformees = transformer(donnees_brutes)
    charger(donnees_transformees)

# Instanciation obligatoire du DAG
pipeline_covid_taskflow()

[!TIP] Quand utiliser la TaskFlow API plutôt que les Opérateurs ? Utilisez la TaskFlow API (@task) pour tous vos traitements Python personnalisés où les étapes s'échangent des variables et métadonnées. Utilisez les opérateurs classiques (SQLExecuteQueryOperator, S3CreateBucketOperator, SparkSubmitOperator) pour interagir avec des services tiers qui disposent déjà d'un connecteur optimisé.

---

4. Vérification et exécution dans l'interface

1. Dès l'enregistrement du fichier .py dans votre dossier local dags/, le processus airflow-scheduler détecte le nouveau pipeline en moins de 30 secondes.
2. Ouvrez http://localhost:8080/ : le DAG etl_covid19_daily apparaît dans la liste.
3. Cliquez sur le bouton à bascule à gauche pour l'activer (Unpause).
4. Cliquez sur le bouton d'action Play puis sélectionnez Trigger DAG.
5. Ouvrez le DAG pour observer le passage successif des tâches du statut gris (queued) au vert clair (running), puis au vert foncé (success).

---

Récapitulatif

• Le bloc with DAG(...) as dag: garantit une déclaration propre et lisible sans répéter le paramètre dag=dag dans chaque tâche.
• L'opérateur binaire >> définit la séquence d'exécution logique.
• La TaskFlow API (@task) représente le futur du développement sous Airflow pour le code Python natif.

---

Votre pipeline tourne avec succès ! Dans la prochaine leçon, nous allons approfondir les règles de planification pour comprendre exactement quand et comment vos DAGs se déclenchent.

Partagez sur les réseaux sociaux

Commentaires