監控、警示與回呼

Python 中的 Apache Airflow 入門

Mike Metzger

Data Engineer

Dag 生命週期

每次 Dag 執行與任務都會歷經一連串狀態
- 佇列中（Queued）
- 執行中（Running）
- 最終狀態：成功、失敗、略過
Airflow 會追蹤狀態與轉換

Dag 執行生命週期：從佇列到執行再到最終狀態的流程圖

回呼（Callbacks）

定義在 Dag 執行或任務上的函式，Airflow 會自動呼叫
特定狀態轉換的回呼
- on_failure_callback－Dag 執行／任務失敗
- on_success_callback－Dag 執行／任務成功
任務層級回呼
- on_retry_callback
- on_skipped_callback
- on_execute_callback

Airflow 在狀態轉換時觸發回呼的示意圖

回呼的 context

Airflow 會自動傳入 context 字典
context 包含該次 Dag 執行／任務的資訊
- context["dag"].dag_id－Dag 名稱
- context["task_instance"].task_id－任務名稱
- context["logical_date"]－Dag 執行的日期
回呼函式必須接受 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.")

回呼範例

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（通知器）

可綁定到回呼的函式
對外部系統發送警示
可用的 notifier：
- SmtpNotifier－寄送電子郵件警示
- SlackNotifier－發送訊息到 Slack 頻道
- 特定服務的 notifier－PagerDuty、OpsGenie

Airflow notifier 對外部系統發送警示的示意圖

SmtpNotifier

在回呼時寄出電子郵件
位於 airflow.providers.smtp.notifications.smtp 函式庫
需要 from_email 與 to 屬性

可加入 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（稽核記錄）

使用 Audit Log 監控 Airflow
包含此 Airflow 執行個體的所有事件序列

Airflow 稽核記錄頁面，依事件類型列出系統事件

一起來練習吧！

Python 中的 Apache Airflow 入門