트리거 규칙 구현하기
워크플로를 생성한 후, 하나 이상의 태스크가 실패했을 때 알림을 제공하도록 DAG을 개선하면 좋겠다는 것을 깨달았습니다. 이를 위해 one failed 조건을 확인하는 태스크를 구현하여, DAG 내 어떤 태스크가 실패하더라도 알림을 받을 수 있도록 설정하기로 했습니다.
다른 모든 태스크는 이미 정의되어 있으며, task와 dag 객체도 미리 임포트되어 있습니다.
이 연습은 강의의 일부입니다
Python으로 배우는 Apache Airflow 입문
연습 안내
- 트리거 규칙을 사용하기 위한 적절한 라이브러리를 임포트하세요.
notify_on_failure태스크에 적절한 트리거 규칙 속성을 추가하세요.- 하나 이상의 업스트림 태스크가 실패할 때 해당 태스크가 트리거되도록 속성을 설정하세요.
notify_on_failure를 두 개의 변환 태스크의 다운스트림 의존성으로 설정하세요.
실습형 인터랙티브 연습
이 예제를 이 샘플 코드를 완성하여 풀어보세요.
# 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()