Dans un système de données complexe, l'ordre et les conditions d'exécution sont primordiaux. Une transformation ne doit pas démarrer si l'ingestion a échoué, une alerte d'erreur doit se déclencher uniquement si une anomalie est détectée, et un rapport d'entreprise ne doit être généré que lorsque trois pipelines distincts ont tous terminé leurs calculs.
Dans cette leçon, nous allons explorer les techniques avancées pour orchestrer vos flux au sein d'un même DAG (Intra-DAG) et entre différents DAGs (Inter-DAG).
1. Dépendances de base au sein d'un DAG (Intra-DAG)#
Airflow surcharge les opérateurs Python de décalage de bits (Bitshift operators) pour définir le flux d'exécution :
- >> (Right bitshift) : La tâche de gauche s'exécute avant celle de droite (Downstream).
- << (Left bitshift) : La tâche de droite s'exécute avant celle de gauche (Upstream).
# La tâche B attend la fin avec succès de la tâche A
tache_a >> tache_b
# Équivalent syntaxique (à éviter car moins intuitif à lire de gauche à droite) :
tache_b << tache_a
Parallélisme et listes de tâches#
Vous pouvez facilement paralléliser plusieurs traitements indépendants en reliant des listes de tâches :
graph LR
A["tache_a<br><i>(Ingestion)</i>"] --> B["tache_b<br><i>(Calcul Ventes)</i>"]
A --> C["tache_c<br><i>(Calcul Stock)</i>"]
B --> D["tache_d<br><i>(Agrégation Finale)</i>"]
C --> D
style A fill:#f1f5f9,stroke:#475569
style B fill:#e0f2fe,stroke:#0284c7
style C fill:#e0f2fe,stroke:#0284c7
style D fill:#dcfce7,stroke:#16a34a,stroke-width:2px,color:#0f172a
# Tâche A démarre B et C en parallèle ; D attend la fin conjointe de B et C
tache_a >> [tache_b, tache_c] >> tache_d
Les utilitaires chain et cross_downstream#
Lorsque vous avez des matrices de dépendances complexes, l'utilisation de chain améliore considérablement la lisibilité :
from airflow.models.baseoperator import chain
# Équivaut à t1 >> t2 >> t3 >> t4
chain(t1, t2, t3, t4)
2. L'aiguillage conditionnel (Branching)#
Toutes les exécutions ne suivent pas un chemin fixe. Souvent, vous devez inspecter les données et choisir dynamiquement la suite du parcours :
flowchart TD
CHECK["verifier_qualite_donnees"] --> DECISION{"Score Qualité > 95% ?"}
DECISION -->|Oui| PROD["charger_en_production<br><i>(Exécutée)</i>"]
DECISION -->|Non| QUARANT["isoler_en_quarantaine<br><i>(Exécutée)</i>"]
PROD --> JOIN["cloture_pipeline"]
QUARANT --> JOIN
style CHECK fill:#f1f5f9,stroke:#475569
style DECISION fill:#fef3c7,stroke:#d97706,stroke-width:2px,color:#0f172a
style PROD fill:#dcfce7,stroke:#16a34a
style QUARANT fill:#fee2e2,stroke:#ef4444
style JOIN fill:#e0f2fe,stroke:#0284c7
Avec la TaskFlow API (@task.branch)#
La méthode moderne consiste à utiliser le décorateur @task.branch. La fonction renvoie simplement le task_id (ou une liste de task_ids) de la tâche à exécuter :
from airflow.decorators import task
@task.branch
def orienter_flux(**context) -> str:
# Récupération d'un métrique ou résultat de test
nb_lignes_rejetees = 12
if nb_lignes_rejetees == 0:
return "charger_en_production"
else:
return "isoler_en_quarantaine"
Les tâches non choisies par le branchement sont automatiquement marquées avec le statut Skipped (Rose) dans l'interface, et leurs enfants sont également ignorés par défaut.
3. Les règles de déclenchement (Trigger Rules)#
Par défaut, une tâche ne démarre que si toutes ses tâches parentes directes ont réussi (all_success).
Mais si vous avez un aiguillage conditionnel (où une des branches est ignorée), la tâche de clôture en aval sera bloquée car l'une de ses branches parentes a été passée sous silence (skipped) !
Pour résoudre cela, vous devez ajuster le paramètre trigger_rule de la tâche :
| Trigger Rule | Définition & Comportement | Cas d'usage type |
|---|---|---|
all_success (Défaut) |
Toutes les tâches en amont doivent être en statut success. | Enchaînement séquentiel strict standard. |
none_failed_min_one_success |
Aucune tâche amont n'a échoué et au moins une a réussi (ignore les branches skipped). | Point de ralliement idéal après un aiguillage par Branching. |
all_done |
Démarre dès que toutes les tâches amont sont terminées, peu importe qu'elles aient réussi ou échoué. | Nettoyage de ressources : suppression de clusters temporaires, fermeture de sessions. |
one_failed |
Démarre immédiatement dès qu'au moins une tâche amont échoue. | Déclenchement d'une alerte d'incident immédiate sans attendre la fin des autres tâches. |
all_failed |
Toutes les tâches amont ont échoué. | Procédure de rollback d'urgence globale. |
always |
S'exécute quoi qu'il arrive en amont sans aucune condition. | Logs de monitoring final. |
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule
cloture = EmptyOperator(
task_id="cloture_pipeline",
trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
)
4. Dépendances entre DAGs distincts (Inter-DAG)#
Dans les architectures data professionnelles, les pipelines sont découpés en micro-DAGs modulaires :
- dag_ingestion_crm (géré par l'équipe Data Ops)
- dag_datamart_marketing (géré par l'équipe Data Analytics)
Comment faire en sorte que le deuxième démarre dès la fin du premier ?
Méthode 1 : Les Datasets (Recommandée depuis Airflow 2.4)#
C'est la méthode la plus élégante : le premier DAG déclare écrire dans une ressource, et le second écoute cette ressource. Aucun couplage rigide !
# Dans dag_ingestion.py :
tache_ecriture = EmptyOperator(
task_id="ecrire_donnees",
outlets=[Dataset("postgres://dw/crm_contacts")]
)
# Dans dag_analytics.py :
with DAG(
dag_id="dag_analytics",
schedule=[Dataset("postgres://dw/crm_contacts")],
...
):
...
Méthode 2 : L'ExternalTaskSensor#
Si vous utilisez une version antérieure ou si vous devez attendre une tâche spécifique dans un DAG externe :
from datetime import timedelta
from airflow.sensors.external_task import ExternalTaskSensor
attente_crm = ExternalTaskSensor(
task_id="attendre_ingestion_crm",
external_dag_id="dag_ingestion_crm",
external_task_id="fin_ingestion",
timeout=3600,
mode="reschedule", # TRÈS IMPORTANT : libère le worker entre chaque vérification !
poke_interval=60, # Vérifie l'état toutes les 60 secondes
)
[!WARNING] Règle vitale pour les Sensors : utilisez toujours
mode="reschedule"! Le mode par défaut (mode="poke") monopolise un slot de worker pendant toute la durée d'attente. Si vous avez 5 sensors en mode poke qui attendent pendant 2 heures, ils occuperont tous vos workers et paralyseront l'intégralité de votre cluster Airflow ! Le mode reschedule libère le conteneur entre chaque tentative de vérification.
Méthode 3 : Le TriggerDagRunOperator#
Permet à un DAG parent de commander directement l'exécution d'un DAG enfant :
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
declencher_reporting = TriggerDagRunOperator(
task_id="lancer_reporting",
trigger_dag_id="dag_datamart_marketing",
wait_for_completion=False, # Poursuit sans attendre que le DAG enfant finisse
)
Vous possédez désormais tout l'arsenal pour orchestrer des workflows sophistiqués et robustes. Dans la prochaine leçon, nous allons explorer la CLI d'Airflow pour piloter vos déploiements en ligne de commande !
Commentaires