Triển khai hàm callback
Gần đây bạn được giao nhiệm vụ thêm các failure callback cho các Dag do đội của bạn tạo. Để bắt đầu, bạn muốn thêm một failure callback đơn giản để ghi một thông điệp vào audit log khi sales_etl_dag thất bại.
Các đối tượng dag và task đã được import và các task get_sales_data và process_sales_data đã được tạo.
Bài tập này là một phần của khóa học
Giới thiệu về Apache Airflow bằng Python
Hướng dẫn bài tập
- Tạo một hàm callback tên
alert_on_failure. - Định nghĩa hàm để chấp nhận bất kỳ đối tượng nào Airflow truyền vào.
- Chỉ định một failure callback sử dụng hàm
alert_on_failure.
Bài tập tương tác thực hành trực tiếp
Hãy thử làm bài tập này bằng cách hoàn thành đoạn mã mẫu này.
# 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()