Implémenter une règle de déclenchement

Après avoir créé un flux de travail, vous réalisez que le Dag gagnerait à envoyer des mises à jour lorsqu'au moins une tâche échoue. Vous décidez d'ajouter une tâche qui applique la vérification one failed sur votre Dag pour vous alerter si une tâche du Dag échoue.

Toutes les autres tâches ont été définies et les objets task et dag sont déjà importés pour vous.

Cette activité fait partie du cours

Introduction à Apache Airflow en Python

Voir le cours

Instructions de l’exercice

Importez la bibliothèque appropriée pour utiliser les règles de déclenchement (trigger rules).
Ajoutez l'attribut de règle de déclenchement approprié à la tâche notify_on_failure.
Réglez l'attribut pour que la tâche se déclenche lorsqu'une ou plusieurs tâches en amont échouent.
Définissez notify_on_failure comme dépendance en aval des deux tâches de transformation.

Exercice interactif pratique

Essayez cet exercice en complétant ce code d’exemple.

# Import TriggerRule
from airflow.utils.____ import ____

@dag(schedule="@daily", start_date=datetime(2026, 5, 1))
def etl_pipeline():

    # Trigger notify_on_failure when any upstream task fails
    @task(____=TriggerRule.____)
    def notify_on_failure(**context) -> None:
        dag_id = context["dag"].dag_id
        run_id = context["run_id"]
        print(f"ALERT: A task failed in DAG '{dag_id}', run '{run_id}'. Sending notification...")

    # Set notify_on_failure downstream of both transform tasks
    [transform_users(), transform_orders()] ____ notify_on_failure()
    
etl_pipeline()

Modifier et exécuter le code