未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

监控告警与故障处理

---

Web UI 使用指南

DAGs 列表页

登录后进入的主页，核心操作：

• 暂停/恢复 DAG：点击 DAG 名称左侧的开关
• 手动触发：点击 "Trigger DAG" 按钮（三角形图标）
• 删除 DAG：在 DAG 列表找到 DAG，点击垃圾桶图标（仅删除元数据，不删文件）
• 查看运行状态：彩色小方块代表最近几次运行状态（绿=成功，红=失败）

DAG 详情页：四个核心视图

Grid View（网格视图）：

• 横轴是时间（最近 N 次运行），纵轴是 Task 列表
• 快速看到哪些日期哪些 Task 失败了
• 点击某个格子可以看该 Task Instance 的详情

Graph View（图形视图）：

• 显示 DAG 的依赖关系图
• 可以选择某次 DAG Run，查看各 Task 的实时状态
• 点击任意 Task 可以查看日志、重新运行、标记成功等

Calendar View（日历视图）：

• 按月日历显示每天的 DAG Run 状态
• 快速定位哪天出了问题

Gantt View（甘特图视图）：

• 展示每个 Task 的开始和结束时间
• 用于分析任务耗时和并行度是否合理

常用操作

在 Graph View 点击某个 Task 后，可以：
- View Log：查看运行日志
- Run：在支持该操作的版本中请求运行所选 Task Instance；是否执行及执行范围取决于依赖检查和所选选项，任务输出也可能影响下游
- Clear：清除状态，等待 Scheduler 重新调度
- Mark Success：手动标记为成功，不执行任务逻辑；可能使下游满足触发条件，但不保证下游一定运行，使用前必须确认所需数据已就绪
- Mark Failed：强制标记为失败

---

任务重试策略

基础重试配置

from datetime import timedelta

default_args = {
    # 失败后最多重试 3 次
    "retries": 3,
    # 每次重试等待 5 分钟
    "retry_delay": timedelta(minutes=5),
    # 指数退避：等待时间随重试次数增长；实际间隔可能含抖动，并受版本实现和 max_retry_delay 限制，不保证恰为 5、10、20 分钟
    "retry_exponential_backoff": True,
    # 最大重试等待时间（配合指数退避使用）
    "max_retry_delay": timedelta(hours=1),
}

with DAG(
    dag_id="resilient_etl",
    default_args=default_args,
    # ...
) as dag:
    pass

针对不同 Task 设置不同重试策略

# 网络请求容易偶发失败，多重试几次
api_task = PythonOperator(
    task_id="call_external_api",
    python_callable=call_api,
    retries=5,
    retry_delay=timedelta(minutes=2),
    retry_exponential_backoff=True,
)

# Spark 作业比较稳，失败通常是真实问题
spark_task = SparkSubmitOperator(
    task_id="run_spark_job",
    application="/jobs/etl.py",
    retries=1,              # 只重试 1 次
    retry_delay=timedelta(minutes=30),  # 等 30 分钟再重试（让集群资源释放）
)

# 数据质量检查失败不应该重试（失败本身就是有意义的信号）
quality_check = PythonOperator(
    task_id="data_quality_check",
    python_callable=check_quality,
    retries=0,  # 不重试
)

---

on_failure_callback：自定义失败处理

当 Task 失败时，可以执行自定义的回调函数，比如发送告警通知。

基础用法

def task_failure_alert(context):
    """Task 失败时的回调函数"""
    task_instance = context["task_instance"]
    dag_id = context["dag"].dag_id
    task_id = task_instance.task_id
    execution_date = context["execution_date"]
    log_url = task_instance.log_url

    message = f"""
Airflow Task 失败告警

DAG: {dag_id}
Task: {task_id}
执行时间: {execution_date}
查看日志: {log_url}
    """.strip()

    print(f"发送告警：{message}")
    # 这里调用你的告警通知函数

def dag_failure_alert(context):
    """DAG 级别的失败回调（整个 DAG Run 失败时触发）"""
    pass

with DAG(
    dag_id="monitored_dag",
    on_failure_callback=dag_failure_alert,  # DAG 级别失败回调
    # ...
) as dag:

    critical_task = PythonOperator(
        task_id="critical_task",
        python_callable=run_critical,
        on_failure_callback=task_failure_alert,   # Task 级别失败回调
        on_success_callback=task_success_log,      # 成功回调
        on_retry_callback=task_retry_alert,        # 重试回调
    )

---

钉钉告警集成

本文以钉钉群机器人 Webhook 演示一种告警接入方式；是否适用取决于组织使用的平台和机器人安全配置。

import requests
import json
from datetime import datetime

def send_dingtalk_alert(context):
    """发送钉钉群告警"""
    task_instance = context["task_instance"]
    dag_id = context["dag"].dag_id
    task_id = task_instance.task_id
    execution_date = str(context["execution_date"])
    log_url = task_instance.log_url

    # 从环境变量读取钉钉 Webhook URL（不要硬写在代码里）
    import os
    webhook_url = os.environ.get("DINGTALK_WEBHOOK_URL")
    if not webhook_url:
        print("DINGTALK_WEBHOOK_URL 未配置，跳过钉钉告警")
        return

    # 构建钉钉消息（Markdown 格式）
    message = {
        "msgtype": "markdown",
        "markdown": {
            "title": f"Airflow 告警：{dag_id}",
            "text": f"""## Airflow Task 失败告警
**DAG**: {dag_id}
**Task**: {task_id}
**执行日期**: {execution_date}
**失败时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

[查看日志]({log_url})
""",
        },
        "at": {
            "isAtAll": False,
        },
    }

    # 发送告警
    try:
        response = requests.post(
            webhook_url,
            headers={"Content-Type": "application/json"},
            data=json.dumps(message),
            timeout=10,
        )
        if response.json().get("errcode") == 0:
            print("钉钉告警发送成功")
        else:
            print(f"钉钉告警发送失败：{response.text}")
    except Exception as e:
        print(f"钉钉告警发送异常：{e}")

# 在 DAG 中使用
with DAG(
    dag_id="production_etl",
    default_args={
        "on_failure_callback": send_dingtalk_alert,
        "retries": 2,
    },
    # ...
) as dag:
    pass

---

飞书告警集成

飞书机器人的接入方式与钉钉类似：

def send_feishu_alert(context):
    """发送飞书群告警"""
    import requests
    import json
    import os

    task_instance = context["task_instance"]
    dag_id = context["dag"].dag_id
    task_id = task_instance.task_id
    execution_date = str(context["execution_date"])
    log_url = task_instance.log_url

    webhook_url = os.environ.get("FEISHU_WEBHOOK_URL")
    if not webhook_url:
        return

    # 飞书消息格式（Rich Text）
    message = {
        "msg_type": "post",
        "content": {
            "post": {
                "zh_cn": {
                    "title": f"Airflow 告警：{dag_id}",
                    "content": [
                        [{"tag": "text", "text": f"DAG: {dag_id}"}],
                        [{"tag": "text", "text": f"Task: {task_id}"}],
                        [{"tag": "text", "text": f"执行日期: {execution_date}"}],
                        [
                            {"tag": "text", "text": "查看日志："},
                            {"tag": "a", "text": "点击这里", "href": log_url},
                        ],
                    ],
                }
            }
        },
    }

    try:
        response = requests.post(webhook_url, json=message, timeout=10)
        response.raise_for_status()
        print(f"飞书告警请求已提交，须按当前接口约定检查业务响应并验证实际送达：{response.text}")
    except Exception as e:
        print(f"飞书告警异常：{e}")

---

邮件告警配置

Airflow 内置支持邮件告警，通过配置 SMTP 即可：

# airflow.cfg 或环境变量
[smtp]
smtp_host = smtp.exmail.qq.com        # 企业邮箱 SMTP 地址
smtp_starttls = True
smtp_ssl = False
smtp_user = airflow-alert@company.com
smtp_password = your_email_password
smtp_port = 587
smtp_mail_from = airflow-alert@company.com

# 在 default_args 中启用邮件告警
default_args = {
    "email": ["data-team@company.com"],
    "email_on_failure": True,   # 失败时发邮件
    "email_on_retry": False,    # 重试时不发（避免邮件轰炸）
    # 标准任务参数不提供 email_on_success；如需成功通知，可在 on_success_callback 中显式发送
}

---

日志查看与排查技巧

Web UI 查看日志

1. 进入 DAG 的 Grid View 或 Graph View
2. 点击失败的 Task（红色方块）
3. 点击 "View Log"
4. 每次运行会有单独的日志文件（attempt_1.log、attempt_2.log...）

命令行查看日志

# 查看特定 Task Instance 的日志：使用 Web UI 的 View Log 或部署配置的日志后端
# 不假定存在 airflow tasks logs 子命令；先查看当前版本支持的任务命令
airflow tasks --help

# 查看最近的错误日志
tail -100 $AIRFLOW_HOME/logs/scheduler/latest/my_dag.py.log

# Docker 环境下
docker compose logs -f airflow-scheduler | grep -i "error\|exception"
docker compose logs -f airflow-worker | tail -100

常见日志模式

# 任务找不到 Python 模块
ModuleNotFoundError: No module named 'pandas'
# 解决：在 Airflow 环境中安装缺少的包

# 数据库连接失败
sqlalchemy.exc.OperationalError: (mysql.connector.errors.InterfaceError)
# 解决：检查 Connection 配置，验证数据库是否可达

# XCom 数据太大
sqlalchemy.exc.DataError: (1406, "Data too long for column 'value'")
# 解决：XCom 不适合传大数据，改用文件路径传递

# 任务僵尸（zombie task）
[2024-01-15 02:30:00] {taskinstance.py:1} WARNING - Detected zombie job
# 解决：先确认原任务进程已停止、外部作业状态及写入幂等性，再清除对应的 Task Instance 并重新调度

---

常见故障处理手册

故障 1：DAG 不出现在 Web UI

症状：把 DAG 文件放到 dags 目录后，Web UI 里看不到

排查步骤：

# 1. 检查 DAG 文件是否有语法错误
python dags/my_dag.py

# 2. 查看 Airflow 导入错误
airflow dags list-import-errors

# 3. 检查 Scheduler 日志
docker compose logs airflow-scheduler | grep "ERROR\|my_dag"

# 4. 在支持该命令的版本中重新生成序列化 DAG；这不等同于触发 Scheduler 立即扫描
# 操作前先确认当前版本的命令及参数
airflow dags reserialize --help

常见原因：

• Python 语法错误
• 缺少 import 的模块
• 循环导入
• DAG 实例未被当前版本的发现机制识别：检查是否通过模块全局变量暴露 DAG 实例，或使用该版本支持的自动注册方式；变量名不必为 dag，也不必须使用 with DAG(...) as dag:

故障 2：任务一直是 queued 状态

症状：任务应该运行，但状态停在 queued 很久

排查步骤：

# 检查 Worker 是否在线
# 不要用 airflow celery worker --pid 查看进程；worker 子命令用于启动 Worker
# 在部署平台的进程或容器列表中检查已有 Worker
# 或
docker compose ps | grep worker

# 检查 Worker 并发是否已满：结合 Worker 配置、日志及执行器监控查看活动任务数
# 另检查 Pool slot 是否耗尽：在提供该菜单的版本中，通过 Web UI → Admin → Pools 查看任务所属 Pool 的使用情况

# 检查任务队列：在 CeleryExecutor 部署中使用已配置的 Celery 应用执行 inspect active
# Airflow CLI 不保证提供 airflow celery inspect 子命令；先确认当前版本支持的命令
airflow celery --help

常见原因：

• Worker 进程没有启动或崩溃
• Worker 并发数（worker_concurrency）设置太小
• Pool slot 耗尽

故障 3：任务成功但下游没有触发

症状：某个 Task 显示成功，但依赖它的下游 Task 没有运行

排查步骤：

# 检查下游 Task 的 trigger_rule 配置
# 检查下游 Task 的上游是否存在 BranchPythonOperator，以及未选中的分支是否被 skip
# 默认 trigger_rule="all_success" 要求所有直接上游成功；直接上游存在 skipped 时不满足该条件

# 检查 DAG 的 depends_on_past 配置
# 如果上一次这个 Task 失败了，本次可能被阻塞

故障 4：Scheduler 没有触发应该运行的 DAG

症状：cron 时间到了，但 DAG 没有新的 DAG Run

排查步骤：

# 1. 检查 DAG 是否被暂停
airflow dags list | grep my_dag
# is_paused 应该为 False

# 2. 检查 start_date 是否比当前时间晚
# start_date 设置了未来的日期？

# 3. 检查 max_active_runs 是否已满
# 已有 DAG Run 在执行，达到 max_active_runs 限制？

# 4. 检查 Scheduler 是否运行正常
# 不要假定 scheduler 提供 --health-check；使用目标版本支持的健康检查方式
# 支持 jobs check 的 Airflow 2.x 部署可检查本机 SchedulerJob 心跳
airflow jobs check --job-type SchedulerJob --local

故障 5：僵尸任务（Zombie Task）

症状：任务明明已经结束（Worker 进程消失），但状态还显示 running

原因：Worker 意外宕机，来不及更新 MetaDB 中的状态

解决方案：

# 在 Web UI 中：
# 找到对应的 Task Instance → 右键 → Mark Failed
# 然后 Clear 重新调度

# 命令行清除
airflow tasks clear my_dag -t stuck_task -s 2024-01-15 -e 2024-01-15

# 配置 Scheduler 的僵尸任务检测（airflow.cfg）
# [scheduler]
# scheduler_zombie_task_threshold = 300  # 适用于提供该配置项的 Airflow 2.x；表示判定僵尸任务的心跳阈值，实际检测还受检查周期影响

---

告警通知最佳实践

告警场景	推荐方式	配置位置
关键任务失败	钉钉/飞书 + 邮件	on_failure_callback
任务重试	仅记录日志（避免轰炸）	email_on_retry=False
SLA 超时	钉钉/飞书	sla_miss_callback
任务成功	仅在最后汇总任务发送	on_success_callback（最后一个 Task）
整个 DAG 失败	高优先级告警	DAG 的 on_failure_callback

---

小结

• Web UI 的 Grid View 是日常监控的核心，快速定位失败的 Task
• 重试策略要区分场景：网络请求多重试，数据质量检查不重试
• 告警用 on_failure_callback，国内推荐钉钉/飞书 Webhook
• 排查问题的第一步永远是 查日志：Web UI 的 View Log 或 Scheduler 日志
• 僵尸任务处理：先确认原任务及外部作业已停止、重跑不会重复写入，再按实际状态使用 Mark Failed 或 Clear 重新调度

下一篇：10-数据管道实战 — 完整端到端案例：MySQL → 清洗 → 数仓 → 报表 → 邮件

---

所属系列：00-Airflow学习路线图 | 上一篇：08-变量连接与配置管理 | 下一篇：10-数据管道实战

---

---

掌握检查

• [ ] 能写出一个带钉钉 Webhook 告警的 on_failure_callback 函数，并在 DAG 的 default_args 中注册它
• [ ] 能区分 retries（重试次数）、retry_delay（重试间隔）、execution_timeout（执行超时）三个参数的作用
• [ ] 知道"僵尸任务"的成因（Worker 宕机未更新 MetaDB），以及如何通过 Web UI 或命令行清除
• [ ] 能说出排查 DAG 不触发问题的完整步骤（至少 4 步：检查开关、检查 start_date、检查 max_active_runs、检查 Scheduler 日志）
• [ ] 理解为什么 email_on_retry 应该设为 False，而 email_on_failure 设为 True

---

实践练习

基础题：给你的测试 DAG 配置钉钉告警：在 default_args 中注册 on_failure_callback，函数内用 requests.post 调用钉钉 Webhook，发送包含 DAG ID、Task ID 和失败时间的消息。手动让一个 Task 失败，验证能否收到钉钉消息。

进阶题：生产环境中有一个 Task 经常因为短暂的网络波动失败，重试 2 次后通常会成功。但数据质量检查 Task 一旦失败说明数据本身有问题，不应该重试。如何在同一个 DAG 中为不同的 Task 配置不同的重试策略？写出具体代码。

#Airflow #监控 #告警 #故障排查 #重试策略 #钉钉 #飞书