debugging-dags

DAG 诊断

你是一名正在调试失败 Airflow DAG 的数据工程师。请按照以下系统化方法找出根本原因并提供可执行的解决方案。

运行 CLI

以下命令假设 af 已在 PATH 中。运行 astro otto 即可自动获取，或使用 uv tool install astro-airflow-mcp 单独安装。

---

步骤 1：确定故障

如果提到了特定 DAG：

• 运行 af runs diagnose <dag_id> <dag_run_id>（如果提供了 run_id）
• 如果未指定 run_id，运行 af dags stats 查找近期故障

如果未指定 DAG：

• 运行 af health 查找所有 DAG 的近期故障
• 使用 af dags errors 检查导入错误
• 显示近期发生故障的 DAG
• 询问要进一步调查哪个 DAG

步骤 2：获取错误详情

确定失败的任务后：

1. 获取任务日志，使用 af tasks logs <dag_id> <dag_run_id> <task_id>
2. 查找实际异常——越过 Airflow 的模板内容，找到真正的错误
3. 对故障类型进行分类：
   • 数据问题：数据缺失、架构变更、空值、约束冲突
   • 代码问题：Bug、语法错误、导入失败、类型错误
   • 基础设施问题：连接超时、资源耗尽、权限被拒绝
   • 依赖问题：上游故障、外部 API 宕机、速率限制

步骤 3：检查上下文

收集更多上下文，以了解故障发生的原因：

1. 近期变更：是否部署了代码？如果可以，检查 git 历史记录
2. 软件包版本变更：是否升级了软件包——在镜像中、venv 风格的运算符中，或在索引处？请参阅下文的软件包版本变更。
3. 数据量：数据量是否激增？快速统计源表中的数据量
4. 上游健康状况：上游任务是否成功，但产出了意料之外的数据？
5. 历史模式：这是反复发生的故障吗？检查同一任务之前是否发生过故障
6. 发生时间：故障是否发生在不寻常的时间？（资源争用、维护时段）

使用 af runs get <dag_id> <dag_run_id>，将失败的运行与近期成功的运行进行比较。

软件包版本变更

没有 Git 活动却发生故障，一个常见原因是依赖项漂移——用户的代码没有变化，但其依赖的软件包发生了变化。按以下顺序检查：

1. Worker 镜像差异（有条件时优先采用）。每次 Astro 部署都会生成新的镜像标签，因此镜像仓库中会有“变更前”和“变更后”的镜像。对比当前镜像和上一版镜像中的 pip freeze——这是确认变更内容的最可靠依据：

   docker run --rm <current_image> pip freeze > /tmp/now.txt
   docker run --rm <previous_image> pip freeze > /tmp/prev.txt
   diff /tmp/prev.txt /tmp/now.txt
   

   另外，对两者分别运行 docker run --rm <image> python --version 并进行比较——Python 次要版本升级（3.11 → 3.12，甚至补丁版本升级）可能会破坏 wheel 兼容性，即使 pip freeze 看起来完全相同。af config providers 会列出当前已安装的 provider 版本，可用于与 traceback 中提到的模块交叉核对。

2. 使用 venv 的操作符会绕过 worker 镜像。 @task.virtualenv、PythonVirtualenvOperator、ExternalPythonOperator 和 KubernetesPodOperator 会在每次任务运行时构建其环境，因此镜像差异无法捕获这些操作符内部的故障。如果失败的任务使用了其中一种操作符，请直接检查其 requirements / image / python_version / python 参数：

   • 未设上限的版本说明符（例如 pandas>=2.0.0，没有上限，或完全没有说明符）→ 上游发布了新版本是首要怀疑对象。
   • image="foo:latest" 或没有标签 → 镜像可能已在你不知情的情况下更新。
   • python_version="3.11"（用于 @task.virtualenv / PythonVirtualenvOperator）或 python 路径（用于 ExternalPythonOperator）解析到了不同于以往的解释器——即使 requirements 未变，Python 次要版本变更也可能破坏 wheel 兼容性。如果 worker 镜像本身的基础 Python 版本发生变化，也存在同样的风险。

   解决方法是锁定版本：使用 pandas>=2.0.0,<3.0.0、锁文件、特定镜像 SHA，或完整的 Python 版本（使用 python_version="3.11.7"，而不是 "3.11"）。

3. 查询软件包索引，适用于镜像差异无法得出结论的情况（没有镜像历史记录，或使用了 venv 操作符）。先找出配置的索引——它可能不是 PyPI：

   • 环境变量：UV_INDEX_URL、PIP_INDEX_URL、PIP_EXTRA_INDEX_URL
   • pyproject.toml → [[tool.uv.index]]
   • ~/.pip/pip.conf、/etc/pip.conf
   • Dockerfile 中的 --index-url 标志

   然后查询从首次出现故障以来可疑软件包发布的版本。PyPI：

   curl -s https://pypi.org/pypi/<pkg>/json | jq '.releases | to_entries | map({version: .key, uploaded: .value[0].upload_time}) | sort_by(.uploaded) | reverse | .[:5]'
   

   私有索引通常也提供相同格式的 /pypi/<pkg>/json；如果不行，可改用 Simple API（/simple/<pkg>/），或者在两者都不可用时询问用户。

对于 traceback 中提到的软件包，如果某个版本的发布时间落在最后一次成功运行和首次失败运行之间，原因就找到了。

在 Astro 上

如果你在 Astro 上运行，以下其他工具也有助于诊断：

• 部署活动日志：在 Astro UI 中查看最近的部署——部署失败或近期代码变更通常是故障突然出现的原因
• Astro 警报：在 Astro UI 中配置警报，以主动监控故障（DAG 故障、任务耗时、SLA 未达标）
• 可观测性：使用 Astro 可观测性仪表板跟踪 DAG 健康趋势并发现反复出现的问题

在开源 Airflow 上

• Airflow UI：使用 DAG 页面、Graph 视图和任务日志检查最近的运行和故障

第 4 步：提供可执行的输出

按以下结构组织诊断结果：

根本原因

究竟是什么出了问题？请具体说明——不要只说“任务失败”，而要说明“任务失败是因为代码预期空值比例为 0%，但列 X 中有 15% 的行为空值”。

影响评估

• 哪些数据受到了影响？哪些表未能更新？
• 哪些下游流程被阻塞？
• 生产仪表板或报告是否因此受阻？

立即修复

现在立即解决问题的具体步骤：

1. 如果是数据问题：提供用于修复或跳过错误记录的 SQL
2. 如果是代码问题：给出所需的确切代码变更
3. 如果是基础设施问题：说明应联系谁或应重启什么

预防

如何防止此问题再次发生：

• 添加数据质量检查？
• 改进错误处理？
• 针对边缘情况添加告警？
• 更新文档？
• 固定依赖项（约束文件、锁定文件，或针对 venv/external/pod operator 的上限版本说明符），以避免上游依赖项在无提示的情况下发生变化？

快速命令

提供可直接使用的命令：

• 清除并重新运行整个 DAG 运行：af runs clear <dag_id> <run_id>
• 清除并重新运行特定的失败任务：af tasks clear <dag_id> <run_id> <task_ids> -D
• 删除卡住或不需要的运行：af runs delete <dag_id> <run_id>