實作 Trigger Rule

建立好工作流程後，你發現若能在至少有一個任務失敗時提供更新會更好。你決定在 Dag 中加入一個執行 one failed 檢查的任務，當 Dag 中任一任務失敗時提醒你。

其他任務都已定義好，且 task 與 dag 物件也已為你匯入。

本練習屬於課程

Python 中的 Apache Airflow 入門

檢視課程

練習說明

匯入可用於觸發規則的對應函式庫。
在 notify_on_failure 任務上加入適當的觸發規則屬性。
設定該屬性，使此任務在一個或多個上游任務失敗時觸發。
將 notify_on_failure 設為兩個 transform 任務的下游相依性。

動手互動練習

試著完成這個範例程式碼，體驗一下這個練習。

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

編輯並執行程式碼