migrate-airflow-kestra

将 Airflow DAG 迁移到 Kestra

将 Apache Airflow DAG 文件迁移为可用于生产环境的 Kestra 流，并提供正确的命名空间文件、任务顺序和通过架构验证的 YAML。

适用场景

当请求涉及从 Apache Airflow 迁移到 Kestra 时，使用此技能，包括：

• 将 Airflow DAG（.py）迁移为 Kestra 流 YAML
• 将由 @task 装饰的函数或运算符转换为 Kestra 任务
• 将 Airflow DAG 依赖关系转换为 Kestra 的串行/并行任务结构

所需输入

• Airflow DAG .py 文件的路径（$ARGUMENTS）
• 生成文件的输出目录（默认为 DAG 所在目录下的 kestra-migrate/）
• 目标 Kestra 命名空间（如果未提供，请询问；默认为 company.team）

工作流

第 1 步 — 读取并分析 Airflow DAG

完整读取 DAG 文件。提取以下内容：

1. DAG 元数据：dag_id、schedule、default_args、tags
2. 任务：对于每个由 @task 装饰的函数或运算符，记录：
   • 任务 ID（函数名或 task_id）
   • Python 导入项和依赖项（使用的 pip 包）
   • 是否包含重要业务逻辑（pandas 转换、ML、数据处理、API 调用、多步骤计算）——这些将成为命名空间文件
   • 是否为简单的一行操作——这类操作可以内联
3. 依赖关系：映射 >> 运算符、set_upstream/set_downstream 调用以及直接任务调用，构建有序的依赖关系图
4. 并行性：识别具有相同上游任务且彼此无依赖关系的任务——这些任务必须放入 io.kestra.plugin.core.flow.Parallel 中
5. 数据传递：识别 XCom 推送/拉取或任务间传递的返回值——这些内容将转换为 Kestra 中的 outputFiles → inputFiles

第 2 步 — 获取并验证 Kestra 架构

curl -s https://api.kestra.io/v1/plugins/schemas/flow -o /tmp/kestra_schema.json

提取此工作流所需的定义：

python3 -c "
needed = [
    'io.kestra.plugin.scripts.python.Commands',
    'io.kestra.plugin.core.flow.Parallel',
    'io.kestra.plugin.core.flow.WorkingDirectory',
    'io.kestra.plugin.core.trigger.Schedule',
]
content = open('/tmp/kestra_schema.json').read()
for t in needed:
    idx = content.find(f'\"{ t }\"')
    if idx >= 0:
        print(f'=== {t} ===')
        print(content[idx:idx+3000])
"

在读取并验证架构定义之前，不要生成任何 YAML。

第 3 步 — 创建命名空间文件

对于每个包含重要业务逻辑的任务（参见下方规则），将其 Python 函数主体提取为独立脚本，放在 <output-dir>/scripts/ 下。

命名空间文件规则：

• 每个 Airflow 任务对应一个文件，文件名与任务名相同：<task_id>.py
• 脚本必须自包含：从本地文件（例如 products.json）读取输入，并将输出写入本地文件（例如 category_stats.json）
• 删除所有 Airflow 导入项（from airflow …、from pendulum …）——这些没有对应项
• 将 XCom ti.xcom_push/返回值替换为 json.dump(result, open("output.json", "w"))
• 将 XCom ti.xcom_pull 替换为 json.load(open("input.json"))
• 保留所有业务逻辑、pandas 转换、API 调用和辅助函数

何时使用命名空间文件，何时内联：

情况	决策
任务包含 pandas/numpy/ML 逻辑	命名空间文件
任务在处理数据时发起 HTTP 请求	命名空间文件
任务逻辑超过 10 行	命名空间文件
任务仅执行一次 API 调用或 Shell 命令	在流中内联
任务是单行 BashOperator	在流中内联

第 4 步 — 生成 Kestra 流 YAML

应用以下所有规则。将流写入 <output-dir>/flow.yaml。

---

生成规则

任务结构 — YAML 列表标记（-）必须位于 id 行，type 必须紧跟在下一行

这是最重要的格式规则。tasks: 中的每个任务块（包括 Parallel 中的嵌套任务）都必须遵循以下确切格式：

1. YAML 序列短横线（-）必须位于 id: 行，绝不能放在 type: 或任何其他属性上。
2. type: 始终是第二行，缩进与 id: 对齐（不带短横线）。
3. 其余所有属性（containerImage、namespaceFiles、commands 等）都必须位于 type: 之后，缩进级别与 type: 相同。

# ✅ CORRECT — dash on id, type immediately after, then everything else
- id: fetch_products
  type: io.kestra.plugin.scripts.python.Commands
  containerImage: python:3.11
  namespaceFiles:
    enabled: true
    include:
      - scripts/fetch_products.py
  commands:
    - python scripts/fetch_products.py

# ❌ WRONG — dash on type instead of id
- type: io.kestra.plugin.scripts.python.Commands
  id: fetch_products

# ❌ WRONG — dash on containerImage
- containerImage: python:3.11
  id: fetch_products
  type: io.kestra.plugin.scripts.python.Commands

# ❌ WRONG — dash on namespaceFiles
- namespaceFiles:
    enabled: true
  id: fetch_products
  type: io.kestra.plugin.scripts.python.Commands

# ❌ WRONG — dash on commands
- commands:
    - python scripts/fetch_products.py
  id: fetch_products
  type: io.kestra.plugin.scripts.python.Commands

# ❌ WRONG — dash on dependencies
- dependencies:
    - pandas
  id: fetch_products
  type: io.kestra.plugin.scripts.python.Commands

# ❌ WRONG — type is not the second property
- id: fetch_products
  containerImage: python:3.11
  type: io.kestra.plugin.scripts.python.Commands

# ❌ WRONG — extra blank line between id and type
- id: fetch_products

  type: io.kestra.plugin.scripts.python.Commands

无论任务出现在何处，都适用此规则：顶层 tasks:、Parallel.tasks: 中、WorkingDirectory.tasks: 中，以及 triggers: 块中。没有例外。

容器镜像 — 始终使用 python:3.11

每个 Python 任务（io.kestra.plugin.scripts.python.Commands）都必须指定：

containerImage: python:3.11

绝不能省略此属性。绝不能使用 python:3.13-slim 或任何其他镜像。

命名空间文件 — 重要的 Python 逻辑始终使用

对于由命名空间文件支持的任务，始终声明：

namespaceFiles:
  enabled: true
  include:
    - scripts/<task_id>.py

在 commands 中引用脚本：

commands:
  - python scripts/<task_id>.py

任务间的数据传递

使用 outputFiles 捕获脚本写入的 JSON，并将其存储在 Kestra 内部存储中：

outputFiles:
  - result.json

通过 inputFiles 将此输出注入下游任务：

inputFiles:
  result.json: "{{ outputs.<upstream_task_id>.outputFiles['result.json'] }}"

并行处理 — 对应 Airflow 的扇出/扇入

具有共同上游且彼此没有依赖关系的任务必须封装在 io.kestra.plugin.core.flow.Parallel 中：

- id: compute_analytics
  type: io.kestra.plugin.core.flow.Parallel
  tasks:
    - id: compute_category_stats
      type: io.kestra.plugin.scripts.python.Commands
      containerImage: python:3.11
      ...
    - id: compute_brand_stats
      type: io.kestra.plugin.scripts.python.Commands
      containerImage: python:3.11
      ...

在获取/摄取阶段相互独立的任务也应并行运行。

调度触发器

将 Airflow 的 schedule 映射到 Kestra 的 Schedule 触发器：

Airflow	Kestra cron
@daily	@daily
@hourly	@hourly
0 6 * * *	0 6 * * *
None	完全省略触发器

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "@daily"

DAG 元数据映射

Airflow	Kestra
dag_id	id
default_args.owner	# comment 或标签
tags	labels
description	description
catchup=False	无需对应设置
retries	不在 flow 级别设置；如有需要，在每个任务中处理

架构合规性

• 仅使用获取的架构中明确规定的任务类型和属性
• 绝不自行编造属性名称
• 每个任务块中的属性键必须唯一

密钥和凭据

• 绝不硬编码令牌、密码或 API 密钥
• 使用 {{ secret('SECRET_NAME') }} Pebble 表达式

---

输出结构

<output-dir>/
├── flow.yaml                        # Kestra flow
└── scripts/
    ├── <task_id>.py                 # One file per task with business logic
    └── ...

flow.yaml 骨架

id: <dag_id>
namespace: <namespace>
description: <dag description>

tasks:
  # Stage 1 — parallel fetch
  - id: fetch_data
    type: io.kestra.plugin.core.flow.Parallel
    tasks:
      - id: <task_id>
        type: io.kestra.plugin.scripts.python.Commands
        containerImage: python:3.11
        namespaceFiles:
          enabled: true
          include:
            - scripts/<task_id>.py
        dependencies:
          - <pip-package>
        commands:
          - python scripts/<task_id>.py
        outputFiles:
          - <output>.json

  # Stage 2 — parallel transforms
  - id: compute_analytics
    type: io.kestra.plugin.core.flow.Parallel
    tasks:
      - id: <task_id>
        type: io.kestra.plugin.scripts.python.Commands
        containerImage: python:3.11
        namespaceFiles:
          enabled: true
          include:
            - scripts/<task_id>.py
        inputFiles:
          <input>.json: "{{ outputs.<upstream_id>.outputFiles['<input>.json'] }}"
        dependencies:
          - <pip-package>
        commands:
          - python scripts/<task_id>.py
        outputFiles:
          - <output>.json

  # Stage 3 — sequential summary
  - id: <final_task_id>
    type: io.kestra.plugin.scripts.python.Commands
    containerImage: python:3.11
    namespaceFiles:
      enabled: true
      include:
        - scripts/<final_task_id>.py
    inputFiles:
      <a>.json: "{{ outputs.<task_a>.outputFiles['<a>.json'] }}"
      <b>.json: "{{ outputs.<task_b>.outputFiles['<b>.json'] }}"
    dependencies:
      - <pip-package>
    commands:
      - python scripts/<final_task_id>.py
    outputFiles:
      - <output>.json

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "<mapped-cron>"

---

使用 kestra-ops 部署

生成文件后，如果用户希望部署，请使用 kestra-ops 技能。关于命名空间文件上传的重要说明：始终逐个上传文件，以避免路径嵌套问题：

for f in <output-dir>/scripts/*.py; do
  name=$(basename "$f")
  kestractl nsfiles upload <namespace> "$f" "scripts/$name" --override
done

不要使用目录上传（kestractl nsfiles upload <ns> ./scripts scripts）——这会嵌套目录名称，生成 scripts/scripts/<file>，而不是 scripts/<file>。

---

示例提示词

• “将 Airflow 的 dags/ingest_pipeline.py 迁移到 Kestra，输出到 kestra/”
• “将此 Airflow DAG 转换为 analytics.products 命名空间中的 Kestra flow”
• “迁移位于 dags/etl.py 的 Airflow DAG，并使用 kestra-ops 部署”