监控、告警与回调

Python 中的 Apache Airflow 入门

Mike Metzger

Data Engineer

Dag 生命周期

 

  • 每次 Dag 运行与任务都会经历一系列状态
    • 排队
    • 运行中
    • 终态:成功、失败、跳过
  • Airflow 跟踪状态与转换

Dag 运行生命周期:从排队到运行到终态的示意图

Python 中的 Apache Airflow 入门

回调

  • Airflow 会自动在 Dag 运行或任务上调用的函数
  • 针对特定状态转换的回调
    • on_failure_callback - Dag 运行/任务失败
    • on_success_callback - Dag 运行/任务成功
  • 任务级回调
    • on_retry_callback
    • on_skipped_callback
    • on_execute_callback

Airflow 回调在状态转换时触发的示意图

Python 中的 Apache Airflow 入门

回调上下文

  • 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.")
Python 中的 Apache Airflow 入门

回调示例

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.
Python 中的 Apache Airflow 入门

通知器

  • 可绑定到回调的函数
  • 向外部系统发送告警
  • 可用通知器:
    • SmtpNotifier - 发送邮件告警
    • SlackNotifier - 向 Slack 频道发消息
    • 应用专用通知器 - PagerDuty、OpsGenie

Airflow 通知器向外部系统发送告警的示意图

Python 中的 Apache Airflow 入门

SmtpNotifier

  • 回调时发送邮件
  • 位于 airflow.providers.smtp.notifications.smtp
  • 需提供 from_emailto 属性
  • 可含 subjecthtml_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'
       )
    )
    
Python 中的 Apache Airflow 入门

审计日志

  • 使用审计日志监控 Airflow
  • 记录实例上的全部事件序列

Airflow 审计日志页面按类型列出系统事件

Python 中的 Apache Airflow 入门

Passons à la pratique !

Python 中的 Apache Airflow 入门

Preparing Video For Download...