Implementing a Trigger Rule

After creating a workflow, you realize the Dag would benefit from providing some updates when at least one task fails. You decide to implement a task that implements the one failed check on your Dag to alert you if any task in the Dag fails.

All other tasks have been defined and the task & dag objects are already imported for you.

Diese Übung ist Teil des Kurses

Einführung in Apache Airflow mit Python

Kurs anzeigen

Anleitung zur Übung

Import the appropriate library to use trigger rules.
Add the appropriate trigger rule attribute to the notify_on_failure task.
Set the attribute so the task triggers when one or more upstream tasks fail.
Set notify_on_failure as a downstream dependency of the two transform tasks.

Interaktive Übung

Vervollständige den Beispielcode, um diese Übung erfolgreich abzuschließen.

# 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()

Code bearbeiten und ausführen