migrate-airflow-kestra

Airflow DAG を Kestra に移行する

Apache Airflow の DAG ファイルを、適切な名前空間ファイル、正しいタスク順序、スキーマ検証済み YAML を備えた本番環境対応の Kestra フローに移行します。

使用するタイミング

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 呼び出し、複数ステップの計算）が含まれているかどうか — これらは名前空間ファイルになります
   • 単純な 1 行処理かどうか — これらはインライン化できます
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 タスクごとに 1 ファイル作成し、タスク名を付けます：<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 呼び出しまたはシェルコマンドである	フローにインライン化
タスクが 1 行の BashOperator である	フローにインライン化

ステップ 4 — Kestra フロー YAML を生成する

以下のすべてのルールを適用します。フローを <output-dir>/flow.yaml に書き込みます。

---

生成ルール

タスク構造 — YAML のリストマーカー（-）は必ず id の行に置き、type は必ず次の行に置いてください

これは最も重要な書式ルールです。tasks: 内のすべてのタスクブロック（Parallel 内のネストされたタスクを含む）は、必ず次のパターンに従ってください。

1. YAML のシーケンスダッシュ（-）は id: の行に置きます。type: やほかのプロパティの行には置かないでください。
2. type: は必ず2行目に置き、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 やほかのイメージは使用しないでください。

Namespace ファイル — 重要な Python ロジックには必ず使用してください

Namespace ファイルを使うタスクでは、必ず次を宣言してください。

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	フローレベルでは設定せず、必要に応じてタスクごとに処理

スキーマへの準拠

• 取得したスキーマで明示的に定義されているタスクタイプとプロパティのみを使用してください
• プロパティ名を独自に作らないでください
• 各タスクブロック内でプロパティキーが重複しないようにしてください

シークレットと認証情報

• トークン、パスワード、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 スキルを使用してください。Namespace ファイルのアップロードに関する重要な注意点：パスのネストに関する問題を避けるため、ファイルは必ず個別にアップロードしてください。

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/<file> ではなく scripts/scripts/<file> が生成されます。

---

プロンプトの例

• 「Airflow の dags/ingest_pipeline.py を Kestra に移行し、kestra/ に出力してください」
• 「この Airflow DAG を analytics.products Namespace の Kestra フローに変換してください」
• 「dags/etl.py にある Airflow DAG を移行し、kestra-ops を使ってデプロイしてください」