migrating-dagster-to-airflow

Dagster → Airflow 3（Astro）迁移

将 Dagster 项目迁移到 Astro Runtime 上的 Airflow 3，务必如实。迁移以资产为先（Dagster 资产图转换为 Airflow 资产和资产感知型调度，而不是扁平化的 DAG），循序渐进（按领域逐步迁移，在达到一致性之前，Dagster 始终保持权威地位），并且诚实（每个定义都必须有明确的处置结果；语义差异要记录下来，绝不掩盖）。

第一次进行此迁移？请先阅读 reference/quickstart.md：了解第一小时要用的命令、术语表，以及哪些内容可能出问题、哪些不会。

迁移概览

1. 对源项目运行测试并记录基线，然后以只读方式清点项目（scripts/inventory.py → 清单）。
2. 审查分类（reference/mapping.md 中的 MECH/JUDG/REDESIGN/NONE）；作出是否继续的决定（有三种结果；最常见的是有条件迁移，选择留在原平台的情况较少）；将 DAG 边界、每条边的 IO 决策和 Gate 3 预期纳入清单。
3. 选择 2-3 个有代表性的单元，端到端试迁移，并通过每一道验证关卡。
4. 按领域逐步迁移，遵循六道关卡的流程（reference/validation.md），并跟踪每个单元的状态（scripts/status.py）；依据 reference/troubleshooting.md 修复故障类别，绝不使用存根。
5. 根据 reference/astro-deployment.md 映射平台层（密钥、告警、CI/CD、部署）。
6. 并行运行，然后按领域切换（先取消消费者的暂停；请参阅清单），确保回滚只需一步。
7. 交付迁移报告：为每个定义注明处置结果，为每个触发器提供一行等效性说明，并明确写出损失。

版本差异

已针对 Airflow 3.3.0 / Astro Runtime 3.3-2 / astronomer-cosmos 1.15 / Dagster 1.13（2026-07）完成验证。参考文档中与版本相关的条目标明了最低版本（尤其是 3.2 与 3.3 的分区接口差异）。在依赖版本限制相关声明之前：检查目标环境（airflow version、astro deployment inspect），通过探测导入来验证 SDK 接口（python3 -c "from airflow.sdk import X"），并优先使用 --help / API 规范发现功能，而不要假定较新版本中的 CLI/REST 契约与文档逐字一致。操作手册中的条目均标明了适用版本。

要求

• 目标版本为 Astro Runtime 3.3+（Airflow 3.3+）；原生资产分区接口需要此版本。低于 3.2 时映射会严重退化；请明确说明，并建议先升级再迁移。
• Dagster 仓库，以及最好还要有一个正在运行的 Dagster 实例（其物化元数据可作为等效性测试的样例）。
• 目标项目需要安装 astro CLI。

硬性规则

1. 绝不使用存根。 翻译后的单元要么通过其验证关卡，要么以书面理由延期。伪造成功的函数体，以及附带长篇辩解性注释的变通代码，都不合格。
2. 不得静默遗漏。 清单中的每条记录最终都必须标记为 complete 或 deferred (reason)。否则，scripts/status.py summary 会以非零状态退出；声称完成前必须运行该命令。
3. 为每个触发器提供等效性说明。 每个调度/传感器/自动化条件都要有一行报告：源端写法、目标端写法，以及一句话说明差异。语义差异客观存在（catchup、on_cron 反转、eager 保证）；问题不在于存在差异，而在于没有记录差异。
4. 修复模式，而不是个例。 如果某种转换模式未能通过验证，就修复该模式（并记录到 reference/troubleshooting.md），然后重新应用；不要只对某个单元进行手动修补。
5. 不要臆造 API。 参考文档中只包含已验证的名称。凡是其中未涵盖的内容，使用前都必须对照官方文档验证。

工作流

阶段 0：预检

确认目标 Runtime 版本、astro CLI 是否存在，以及仓库访问权限。识别项目布局：经典布局（@repository/workspace.yaml）、现代布局（Definitions）或 Components 布局（pyproject.toml [tool.dg]、defs.yaml 文件）；三种布局都可能出现，有时还会同时出现。现在就对源项目的测试套件进行基线测试：记录已有失败，并在迁移问题归因中排除这些失败。

阶段 1：清点（只读）

python3 scripts/inventory.py <dagster_repo> --out manifest.json          # static scan
python3 scripts/inventory.py <dagster_repo> --runtime --out manifest.json # + runtime introspection when the project imports

清单列出了每个定义及其文件:行号、捕获的参数、当前与已弃用的拼写，以及带 IO 管理器的依赖边；每条记录都以 classification: "pending" 开头。分类是你的首要判断任务：根据 reference/mapping.md 中对应的行，为每条记录分配 MECH / JUDG / REDESIGN / NONE，并写入清单。扫描器负责枚举（确保结果完整且确定）；Agent 负责分类（作出判断）。无法对应到 mapping.md 某一行的记录本身就是一项发现：请记录下来，不要猜测。此外，还要搜索 DAGSTER_CLOUD_ 和 EnvVar(（平台层，第 5 阶段）。

清单约定：规范清单存放在迁移运行目录中。Astro 项目创建后（第 2 阶段搭建），将清单复制到其 include/inventory/manifest.json，以便 Gate 3 pytest 和 status.py 默认能找到它；在此之前，清单只需保留在运行目录中（之后请保持两份清单同步，以运行目录中的副本为准）。静态记录是规范迁移单元；运行时模式的补充信息会合并到这些记录中，只有确实仅在运行时存在的定义（由工厂生成）才会成为新的单元。

现在生成迁移报告框架：每条清单记录对应一个章节，并加入来自 reference/astro-deployment.md 的密钥/环境变量命名映射。框架的规模应与项目相称：没有密钥的本地项目只需一行说明“无密钥/平台层”，而不是添加空洞的模板章节。

阶段 1.5：继续或停止（诚实的关口）

在翻译任何内容之前，先回答清单足以回答的项目级问题：迁移会让这个团队失去什么？每项损失是否都有可接受的应对方案？ 根据这些因素对本团队的重要程度评估 NONE 和 REDESIGN 行，同时评估缓解方案，而不只是损失本身：

如果对本团队至关重要	Airflow 领域的应对方案	只有在以下情况下才应考虑留下
代码变更时重建 dbt（code_version_changed()）	在 cron 上运行感知状态的 dbt 构建（Fusion / dbt State 每次运行都会跳过未更改的模型，因此部署后的定时任务只会重建实际发生变化的部分），和/或在合并时由 CI 触发 dbt build（由 PR 把关，通常是升级）	团队既无法运行感知状态的 dbt 技术栈，也无法从 CI 运行 dbt
新鲜度驱动的物化	Astro Observe 新鲜度 SLA / Timeliness 告警 + 按 SLA 设定规模的定时运行	定时运行 + 告警确实无法替代新鲜度触发的计算
按资产核算成本（Insights）	Astro Observe 提供管道级别的数据仓库成本管理；会失去按资产粒度的核算	按资产进行成本分摊是合同或组织层面的要求
将资产目录 / 列级血缘作为日常工具	Airflow 3 资产视图 + OpenLineage/Astro 血缘（资产级）	列级血缘已融入日常工作流，且没有外部目录
深度 AutomationCondition 组合、can_subset、按分区选择性物化	大多数情况都可分解为 cron/资产计划（参见 reference/automation.md）；剩余部分则按领域重新设计	多个领域依赖的组合无法分解为任何可行方案
传感器游标事务性、运行范围内的清理	幂等消费者 + max_active_runs；任务主体中的上下文管理器	恰好一次的事件合并是正确性要求，且幂等性无法满足

这张表隐含的一条规则，直白地说就是：仅由 dbt 引起的条件绝不会导致“留下”。 有了 Cosmos、感知状态的 dbt 构建和 CI 触发的构建，每一项 dbt 工作流损失都有符合业内实践的缓解方案（已在本技能的评估计划中通过实际执行验证，包括在真实数据仓库上验证）；dbt 相关事项应记录为条件，而不是阻碍。可观测性相关事项（按资产核算成本、列级血缘）是独立条件，应单独评估，即使团队主要使用 dbt 也不例外。

关口的结果分为三类，其中中间一类最常见：

• 迁移：没有留下的信号；进入阶段 2。
• 有条件迁移（大多数实际项目）：存在损失，且报告第一部分以书面形式列明并接受了相应的缓解方案；特定领域可能需要进行 REDESIGN 工作；记录这些条件后进入阶段 2。
• 今天继续使用 Dagster：仅适用于右栏中多个留下的信号条件确实同时成立，且迁移并非外部强制要求的情况。此时，诚实的交付成果应是书面建议，明确指出未缓解的具体损失，并在此处停止执行。无法说出“不要迁移”的迁移指南，在说“迁移”时也就不可信；但只有无法缓解的损失才能构成“不要迁移”的理由，仅仅存在差异并不足以为此背书。

阶段 2：规划

• DAG 边界：决定哪些资产依赖边应转为资产感知型调度（跨 DAG），哪些应转为任务排序（DAG 内）。按领域、调度频率和团队归属分组；define_asset_job 选择通常会指明自然的领域。
• 逐边确定 IO：通过 reference/io-and-data-passing.md 中的决策树选择（融合／显式存储／XCom）。
• 顺序：先处理叶子领域，再按依赖顺序处理；平台层最后处理。
• 将每个规划单元的目标预期填入清单：为每个单元填写 dag_id、task_count、edges、schedule、asset_outlets。Gate 3 会严格依据这些字段进行断言；缺少这些字段的单元会被验证跳过，因此未补充信息的清单意味着 Gate 3 不会检查任何内容（validate_dag 会明确报告跳过数量，请勿忽略）。
• 搭建目标环境：astro dev init，在 include/ 下放置共享辅助程序。脚手架施加的项目规范（例如要求 retries >= 2 的测试）不得凌驾于源代码的忠实迁移之上：以源代码行为为准；采用项目规范应作为切换后的改进列入报告，并以明确理由跳过脚手架测试。

阶段 3：试迁移

在大规模迁移前，选择 2–3 个有代表性的单元，端到端完成所有关卡。选择一个 MECH 资产、一个分区资产、一个 JUDG 案例，或尽可能接近的组合（小型项目可能没有分区资产或 MECH 资产；此时应选择一个真实 DAG，完整走通一条路径）。试迁移的经验应在扩展前写入 reference/troubleshooting.md；如果试迁移在结构上失败，应停止并重新制定计划，而不是修改迁移单元。

阶段 4：按领域逐一迁移

每个单元的状态机（在清单中跟踪）：

pending → translate → fix-import → fix-lint → fix-tests → verify-parity → complete
                                    ↘ deferred (reason required)

• 根据对应构造的参考文件进行迁移（见下方路由表）。充分的上下文胜过巧妙的技巧：阅读源单元、对应的映射行，以及附近已迁移的示例。
• 通过各关卡进行验证：python3 scripts/validate_dag.py <astro_project> --manifest manifest.json（关卡 1–3），然后按照 reference/validation.md 执行运行验证和一致性验证。
• 关卡失败时，将最新的验证器输出纳入上下文后重试（最多约 10 次，之后应记录失败类别并暂缓处理）。
• 仅在关卡通过后推进状态：python3 scripts/status.py advance <unit-id> ...。如处置状态有误，使用 status.py reopen <unit-id> --reason ... 更正。禁止手动编辑的规则仅适用于 STATE 字段（status）；PLAN 字段（dag_id、task_count、edges、schedule、asset_outlets、target）由规划人员在阶段 2 填写。
• 有意不迁移为独立 DAG 的单元（已并入任务的辅助程序、已转为告警的策略、已转为连接的资源），应标记为 complete，设置 target: "none"，并提供说明其去向的证据；Gate 3 会按设计跳过这些单元。
• 每个单元单独提交，确保提交原子性。

阶段 5：平台层

reference/astro-deployment.md：部署拓扑、密钥／连接命名映射、CI/CD 和预览部署、告警策略映射、Observe／血缘预期，以及 DAGSTER_CLOUD_* 代码内改写检查清单。

阶段 6：并行运行与切换

Dagster 仍为权威系统。以影子运行或暂停状态运行已迁移的 DAG；在相同逻辑时间窗口内比较输出（行数和校验和；根据 Dagster 生成的输出本身重新计算预期值，仅在适当情况下使用记录的物化元数据，详见 reference/validation.md 的 Gate 5）。每个变更窗口只切换一个领域的调度：暂停 Dagster 调度，取消暂停 Airflow DAG；回滚时反向操作。切换后仍应保留 Dagster 中的可读信息（运行历史不会迁移）。

阶段 7：最终报告

scripts/status.py summary 必须通过。报告包含：go/no-go 评估（阶段 1.5）及其理由、每个定义的处置表、所有等价项、明确说明的 NONE/REDESIGN 损失（血缘深度、code_version 触发条件、Insights 成本核算、传感器游标的事务性）、密钥映射，以及列出延期事项及理由的清单。交付前抽查十项 complete 声明。

参考资料路由

遇到的结构	阅读
第一个小时、术语表、可能出现的问题	reference/quickstart.md
任何内容（首先查看：每种结构各占一行）	reference/mapping.md
资产键 → URI 约定、转换粒度、外部/可观测资产	reference/assets.md
资产依赖、IO 管理器、XCom、存储决策	reference/io-and-data-passing.md
任何 partitions_def、分区映射、回填	reference/partitions.md
调度、传感器、AutomationCondition、新鲜度	reference/automation.md
@dbt_assets、转换器、dbt Cloud	reference/dbt.md
组件（defs.yaml）、自定义 Component 子类、动态生成	reference/components.md
dagster_cloud.yaml、密钥、告警、CI/CD、切换	reference/astro-deployment.md
门禁、奇偶校验测试、状态机	reference/validation.md
以往遇到的故障类别	reference/troubleshooting.md

脚本

脚本	用途
scripts/inventory.py	扫描 Dagster 仓库 → JSON 清单（静态模式 + 可选运行时模式）
scripts/validate_dag.py	针对生成的 Astro 项目运行门禁 1-3
scripts/status.py	单元状态机 + 完整性门禁