콜백 함수 구현하기
최근에 팀에서 만든 DAG에 실패 콜백을 추가하는 작업을 맡게 되었습니다. 먼저 sales_etl_dag가 실패했을 때 감사 로그(audit log)에 메시지를 기록하는 간단한 실패 콜백을 추가해 보겠습니다.
dag와 task 객체는 이미 임포트되어 있으며, get_sales_data와 process_sales_data 태스크도 이미 생성되어 있습니다.
이 연습은 강의의 일부입니다
Python으로 배우는 Apache Airflow 입문
연습 안내
alert_on_failure라는 이름의 콜백 함수를 만드세요.- Airflow가 전달하는 모든 객체를 받을 수 있도록 함수를 정의하세요.
alert_on_failure함수를 사용하여 실패 콜백을 지정하세요.
실습형 인터랙티브 연습
이 예제를 이 샘플 코드를 완성하여 풀어보세요.
# Create the callback function
def ____(____):
dag_id = context["dag"].dag_id
task_id = context["task_instance"].task_id
print(f"Task {task_id} in Dag {dag_id} has failed.")
# Specify the Dag with a failure callback
@dag(dag_id='sales_etl_dag',
____=alert_on_failure
)
def sales_etl_dag():
get_sales_data() >> process_sales_data()
sales_etl_dag()