トリガールールの実装
ワークフローを作成した後、少なくとも1つのタスクが失敗した際に通知を送ると便利だと気づきました。そこで、DAG内のいずれかのタスクが失敗した場合に警告を出す one failed チェックを実装したタスクを追加することにします。
他のタスクはすでに定義済みで、task オブジェクトと dag オブジェクトもインポートされています。
この演習はコースの一部です
Python で学ぶ Apache Airflow 入門
演習の手順
- トリガールールを使用するために必要なライブラリをインポートします。
notify_on_failureタスクに適切なトリガールール属性を追加します。- 上流タスクの1つ以上が失敗したときにタスクがトリガーされるよう属性を設定します。
notify_on_failureを2つの変換タスクの下流依存関係として設定します。
実践的なインタラクティブ演習
このサンプルコードを完成させて、この演習に挑戦してみましょう。
# 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()