curiositech/windags-skills

Orchestrateur de DAGs Airflow

Concevez et opérez des DAGs Apache Airflow pour une orchestration fiable de pipelines de données avec une gestion appropriée des dépendances, des SLA et de la surveillance.

Déclencheurs d'activation

Activer sur : "Airflow", "DAG", "opérateur", "capteur", "planificateur", "dépendance de tâche", "SLA", "backfill", "XCom", "API TaskFlow", "MWAA", "Cloud Composer"

PAS pour : exécution de modèles dbt → dbt-analytics-engineer (bien qu'Airflow puisse déclencher dbt) | Traitement de flux → streaming-pipeline-architect | Moteur de workflow (Temporal) → distributed-transaction-manager

Démarrage rapide

1. Définir le DAG — utilisez l'API TaskFlow (@dag, @task decorators) pour des DAGs natifs Python
2. Définir la planification — cron ou timetable, avec catchup=False sauf si le backfill est intentionnel
3. Configurer les tentatives — retries=2, retry_delay=timedelta(minutes=5) sur chaque tâche
4. Ajouter des SLA — sla=timedelta(hours=2) sur les tâches du chemin critique
5. Tester localement — airflow dags test my_dag 2026-01-01 avant le déploiement

Capacités principales

Domaine	Technologies
Airflow	Apache Airflow 2.10+, MWAA, Cloud Composer 3
Opérateurs	BashOperator, PythonOperator, KubernetesPodOperator
Fournisseurs	apache-airflow-providers-{snowflake, google, aws, dbt-cloud}
Exécuteurs	CeleryExecutor, KubernetesExecutor, LocalExecutor
Surveillance	Dépassements de SLA, durée des tâches, métriques Airflow → Prometheus

Modèles d'architecture

DAG avec API TaskFlow

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

@dag(
    schedule="0 6 * * *",          # quotidien à 6h UTC
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=5),
        "sla": timedelta(hours=2),
    },
    tags=["finance", "daily"],
)
def daily_revenue_pipeline():

    @task()
    def extract_payments() -> dict:
        """Extraction depuis l'API Stripe"""
        data = stripe_client.list_payments(date=today())
        return {"count": len(data), "path": "s3://raw/payments/"}

    @task()
    def extract_orders() -> dict:
        """Extraction depuis l'API Shopify"""
        data = shopify_client.list_orders(date=today())
        return {"count": len(data), "path": "s3://raw/orders/"}

    @task()
    def transform(payments: dict, orders: dict) -> str:
        """Jointure et transformation dans DuckDB"""
        result_path = run_duckdb_transform(payments["path"], orders["path"])
        return result_path

    @task()
    def load(path: str):
        """Chargement dans Snowflake"""
        snowflake_copy_into("fct_revenue", path)

    # Définir les dépendances via des appels de fonctions
    payments = extract_payments()
    orders = extract_orders()
    transformed = transform(payments, orders)
    load(transformed)

daily_revenue_pipeline()

Mappage dynamique de tâches (Fan-Out/Fan-In)

@task()
def get_partitions() -> list[str]:
    return ["2026-01-01", "2026-01-02", "2026-01-03"]

@task()
def process_partition(partition_date: str) -> dict:
    """S'exécute en parallèle pour chaque partition"""
    return {"date": partition_date, "rows": process(partition_date)}

@task()
def aggregate(results: list[dict]):
    """Fan-in : reçoit tous les résultats des partitions"""
    total = sum(r["rows"] for r in results)
    log.info(f"Processed {total} total rows")

# Mappe dynamiquement process_partition sur toutes les partitions
partitions = get_partitions()
results = process_partition.expand(partition_date=partitions)
aggregate(results)

Intégration dbt + Airflow

from airflow.operators.bash import BashOperator
from cosmos import DbtDag, ProjectConfig, ProfileConfig

# Option 1 : cosmos (recommandé)
dbt_dag = DbtDag(
    project_config=ProjectConfig("/opt/airflow/dbt/"),
    profile_config=ProfileConfig(
        profile_name="default",
        target_name="prod",
    ),
    schedule="@daily",
    dag_id="dbt_daily",
)

# Option 2 : BashOperator (simple)
dbt_run = BashOperator(
    task_id="dbt_run",
    bash_command="cd /opt/airflow/dbt && dbt build --select tag:daily",
)

Anti-patrons

1. Tâches lourdes — les tâches doivent être atomiques ; ne mettez pas extract+transform+load dans une seule tâche
2. XCom pour de grandes données — XCom est destiné aux métadonnées (chemins, comptages) ; passez les données via S3/GCS, pas via des pickles XCom
3. catchup=True par accident — sauf si vous voulez un backfill, définissez catchup=False ; sinon Airflow exécute chaque intervalle manqué
4. Pas de tentatives — les échecs transitoires sont courants ; définissez toujours retries >= 1 avec un délai
5. Code de niveau supérieur dans les fichiers DAG — les fichiers DAG sont analysés toutes les 30 secondes ; les imports lourds ou les appels API au niveau du module ralentissent le planificateur

Liste de contrôle qualité

• API TaskFlow utilisée pour les DAGs natifs Python (pas d'enchaînement d'opérateurs hérités)
• catchup=False sauf si le backfill est intentionnel
• Tentatives configurées sur toutes les tâches (minimum 1 tentative)
• SLA défini sur les tâches du chemin critique
• XCom passe des références (chemins S3), pas des charges utiles de données
• Mappage dynamique de tâches utilisé pour le parallélisme fan-out
• DAG testé localement avec airflow dags test avant le déploiement
• Idempotence des tâches : réexécuter une tâche produit le même résultat
• Alertes configurées pour les échecs de tâches et les dépassements de SLA
• Tags DAG appliqués pour le filtrage dans l'interface Airflow