authoring-dags

DAG 编写技能

本技能将指导你使用最佳实践和 af CLI 命令创建并验证 Airflow DAG。

如需测试和调试 DAG，请参阅 testing-dags 技能，其中介绍了完整的“测试 -> 调试 -> 修复 -> 重新测试”工作流。

---

运行 CLI

以下命令假设 af 已加入 PATH。运行 astro otto 即可自动使用，也可以通过 uv tool install astro-airflow-mcp 单独安装。

---

工作流概览

+-----------------------------------------+
| 1. DISCOVER                             |
|    Understand codebase & environment    |
+-----------------------------------------+
                 |
+-----------------------------------------+
| 2. PLAN                                 |
|    Propose structure, get approval      |
+-----------------------------------------+
                 |
+-----------------------------------------+
| 3. IMPLEMENT                            |
|    Write DAG following patterns         |
+-----------------------------------------+
                 |
+-----------------------------------------+
| 4. VALIDATE                             |
|    Check import errors, warnings        |
+-----------------------------------------+
                 |
+-----------------------------------------+
| 5. TEST (with user consent)             |
|    Trigger, monitor, check logs         |
+-----------------------------------------+
                 |
+-----------------------------------------+
| 6. ITERATE                              |
|    Fix issues, re-validate              |
+-----------------------------------------+

---

阶段 1：探索

编写代码前，先了解上下文。

探索代码库

使用文件工具查找现有模式：

• 使用 Glob 查找 **/dags/**/*.py，以找到现有 DAG
• 使用 Read 阅读类似的 DAG，了解惯例
• 检查 requirements.txt 中有哪些可用软件包

查询 Airflow 环境

使用 af CLI 命令了解可用内容：

命令	用途
af config connections	已配置了哪些外部系统
af config variables	存在哪些配置值
af config providers	已安装了哪些运算符软件包
af config version	版本限制和功能
af dags list	现有 DAG 和命名惯例
af config pools	用于并发控制的资源池

探索问题示例：

• “是否有 Snowflake 连接？” -> af config connections
• “Airflow 是什么版本？” -> af config version
• “是否提供 S3 运算符？” -> af config providers

---

阶段 2：规划

根据探索结果，提出：

1. DAG 结构 - 任务、依赖关系、调度计划
2. 要使用的运算符 - 根据可用的提供程序选择
3. 所需连接 - 现有连接或需要创建的连接
4. 所需变量 - 现有变量或需要创建的变量
5. 所需软件包 - 添加到 requirements.txt 的软件包

实现前先征得用户同意。

---

阶段 3：实现

按照最佳实践编写 DAG（见下文）。关键步骤：

1. 在合适的位置创建 DAG 文件
2. 如有需要，更新 requirements.txt
3. 保存文件

---

阶段 4：验证

使用 af CLI 作为反馈循环来验证 DAG。

步骤 1：检查导入错误

保存后，检查解析错误（Airflow 已经解析过该文件）：

af dags errors

• 如果出现你的文件 -> 修复后重试
• 如果没有错误 -> 继续

常见原因：缺少导入、语法错误、缺少软件包。

步骤 2：确认 DAG 存在

af dags get <dag_id>

检查：DAG 是否存在、调度计划是否正确、标签是否已设置、暂停状态。

步骤 3：检查警告

af dags warnings

留意弃用警告或配置问题。

步骤 4：查看 DAG 结构

af dags explore <dag_id>

一次调用即可返回：元数据、任务、依赖关系、源代码。

在 Astro 上

如果你在 Astro 上运行，也可以在部署前在本地进行验证：

• 解析检查：运行 astro dev parse，无需启动完整的 Airflow 环境，即可捕获导入错误和 DAG 级别的问题
• 仅部署 DAG：验证通过后，使用 astro deploy --dags 快速仅部署 DAG，无需构建 Docker 镜像，非常适合反复迭代 DAG 代码

---

阶段 5：测试

如需全面的测试指导，请参阅 testing-dags 技能。

验证通过后，按照 testing-dags 技能中的工作流测试 DAG：

1. 征得用户同意 -- 触发前务必先询问
2. 触发并等待 -- af runs trigger-wait <dag_id> --timeout 300
3. 分析结果 -- 检查成功/失败状态
4. 必要时调试 -- af runs diagnose <dag_id> <run_id> 和 af tasks logs <dag_id> <run_id> <task_id>

快速测试（最精简）

# Ask user first, then:
af runs trigger-wait <dag_id> --timeout 300

完整的“测试 -> 调试 -> 修复 -> 重新测试”循环请参阅 testing-dags。

---

阶段 6：迭代

如果发现问题：

1. 修复代码
2. 检查导入错误：af dags errors
3. 重新验证（阶段 4）
4. 按照 testing-dags 技能的工作流重新测试（阶段 5）

---

CLI 快速参考

阶段	命令	用途
发现	af config connections	可用连接
发现	af config variables	配置值
发现	af config providers	已安装的运算符
发现	af config version	版本信息
验证	af dags errors	解析错误（先检查！）
验证	af dags get <dag_id>	验证 DAG 配置
验证	af dags warnings	配置警告
验证	af dags explore <dag_id>	全面检查 DAG

测试命令 -- 请参阅 testing-dags 技能，了解 af runs trigger-wait、af runs diagnose、af tasks logs 等命令。

---

最佳实践与反模式

有关代码模式和反模式，请参阅 reference/best-practices.md。

编写新 DAG 或检查现有 DAG 时，请阅读此参考文档。 其中介绍了哪些模式是正确的（包括 Airflow 3 特有的行为），以及哪些模式应当避免。

---

相关技能

• testing-dags：用于测试 DAG、调试故障，以及执行“测试 -> 修复 -> 重新测试”循环
• debugging-dags：用于排查失败的 DAG
• deploying-airflow：用于将 DAG 部署到生产环境（Astro 或开源版本）
• migrating-airflow-2-to-3：用于将 DAG 迁移到 Airflow 3