Giám sát, Cảnh báo và Callback

Giới thiệu về Apache Airflow bằng Python

Mike Metzger

Data Engineer

Vòng đời Dag

Mỗi lần chạy Dag và mỗi tác vụ đi qua chuỗi trạng thái
- Queued
- Running
- Trạng thái cuối: Success, Failed, Skipped
Airflow theo dõi trạng thái và chuyển tiếp

Sơ đồ vòng đời Dag run từ queued đến running đến trạng thái cuối

Callback

Hàm trên Dag run hoặc Task được Airflow tự động gọi
Callback cho các chuyển trạng thái cụ thể
- on_failure_callback - Dag run / task thất bại
- on_success_callback - Dag run / task thành công
Callback riêng cho Task
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Minh họa callback của Airflow kích hoạt khi chuyển trạng thái

Ngữ cảnh callback

Airflow tự động truyền dictionary context
context chứa thông tin về Dag run / task
- context["dag"].dag_id - Tên Dag
- context["task_instance"].task_id - Tên task
- context["logical_date"] - Ngày chạy Dag
Hàm callback phải nhận context

def alert_on_failure(context):

    dag_id = context["dag"].dag_id
    task_id = context["task_instance"].task_id
    print(f"Task {task_id} in DAG {dag_id} has failed.")

Ví dụ callback

def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["task_instance"].task_id
    print(f"Task {task_id} in Dag {dag_id} has failed.")


@dag(on_failure_callback=alert_on_failure)
def sales_etl_dag():
    @task()
    def data_import_task():
        raise ValueError("Simulated failure")


# Task data_import_task in Dag sales_etl_dag has failed.

Notifier

Các hàm gắn với callback
Gửi cảnh báo đến hệ thống bên ngoài
Nhiều notifier có sẵn:
- SmtpNotifier - Gửi cảnh báo email
- SlackNotifier - Đăng tin vào kênh Slack
- Notifier theo ứng dụng - PagerDuty, OpsGenie

Minh họa notifier của Airflow gửi cảnh báo đến hệ thống bên ngoài

SmtpNotifier

Gửi email khi callback
Trong thư viện airflow.providers.smtp.notifications.smtp
Cần thuộc tính from_email và to

Có thể thêm subject, html_content, ...

@dag(dag_id=`sales_etl_dag`,
   on_failure_callback=SmtpNotifier(
       to='[email protected]',
       from_email='[email protected]',
       subject='Dag sales_etl_dag has failed'
   )
)

Audit Log

Dùng Audit Log để giám sát Airflow
Ghi chuỗi mọi sự kiện trên phiên bản Airflow

Trang nhật ký kiểm toán của Airflow liệt kê sự kiện theo loại

Hãy thực hành!

Giới thiệu về Apache Airflow bằng Python