Een Trigger Rule implementeren

Na het maken van een workflow merk je dat de Dag baat heeft bij updates wanneer minstens één taak faalt. Je besluit een taak te implementeren die de controle one failed op je Dag toepast om je te waarschuwen als een taak in de Dag faalt.

Alle andere taken zijn gedefinieerd en de objecten task & dag zijn al voor je geïmporteerd.

Deze oefening maakt deel uit van de cursus

Introductie tot Apache Airflow in Python

Cursus bekijken

Oefeninstructies

Importeer de juiste bibliotheek om trigger rules te gebruiken.
Voeg het juiste trigger rule-attribuut toe aan de taak notify_on_failure.
Stel het attribuut zo in dat de taak triggert wanneer één of meer upstream-taken falen.
Zet notify_on_failure als downstream-dependency van de twee transform-taken.

Praktische interactieve oefening

Probeer deze oefening eens door deze voorbeeldcode in te vullen.

# 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 bewerken en uitvoeren