migrating-dagster-to-airflow

Dagster → Airflow 3（Astro）移行

Dagster プロジェクトを Airflow 3 on Astro Runtime に移行します。資産を優先し（Dagster の資産グラフは、フラット化した DAG ではなく Airflow の資産と資産対応スケジュールに変換します）、段階的に進め（ドメインごとに移行し、同等性が確認されるまで Dagster を正とします）、誠実に対応します（すべての定義に明示的な処置を割り当て、意味上の差分は隠さず記録します）。

初めて進める場合は、まず reference/quickstart.md をお読みください。最初の1時間で実行するコマンド、用語集、問題が起こり得る箇所と起こり得ない箇所を確認できます。

移行の概要

1. 移行元プロジェクトのテストをベースラインとして記録してから、読み取り専用でインベントリを作成します（scripts/inventory.py → マニフェスト）。
2. 分類（reference/mapping.md に従った MECH/JUDG/REDESIGN/NONE）を確認し、実施可否を判断します（結果は3通りです。一般的なのは条件付きで移行するケースで、移行しない判断は限定的なケースです）。マニフェストに DAG の境界、エッジごとの IO 判断、Gate 3 の想定を計画します。
3. 代表的なユニットを2～3個選び、すべての検証ゲートを通して最初から最後まで試験的に移行します。
4. 6段階のゲート（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"）。また、新しいバージョンで CLI／REST の仕様がそのまま使えると決めつけず、--help や API 仕様の確認を優先してください。プレイブックの各項目には、適用するバージョンの範囲が記載されています。

要件

• 対象は Astro Runtime 3.3+（Airflow 3.3+）です。ネイティブの資産パーティション機能にはこれが必要です。3.2 未満では対応付けの品質が大幅に低下するため、その旨を明示し、移行前のアップグレードを推奨してください。
• Dagster リポジトリ。可能であれば、稼働中の Dagster インスタンスも必要です（そのマテリアライゼーションメタデータを同等性テストのフィクスチャとして利用できます）。
• 対象プロジェクト用の astro CLI。

厳守事項

1. 決してスタブで済ませないでください。 変換したユニットは検証ゲートを通過するか、理由を文書化して延期する必要があります。成功したように見せかける本文や、長い正当化コメント付きの回避策コードは失敗です。
2. 黙って省略しないでください。 インベントリマニフェストの各レコードは、complete または deferred (reason) で終わる必要があります。それ以外の場合、scripts/status.py summary はゼロ以外で終了します。完了を宣言する前に実行してください。
3. すべてのトリガーに同等性の行を設けてください。 各スケジュール／センサー／自動化条件について、レポートに1行を設けます。移行元での表記、移行先での表記、1文での差分を記載してください。意味上の差分（catchup、on_cron の反転、eager の保証）は存在します。問題は差分があることではなく、その差分を文書化しないことです。
4. 個別の事例ではなく、クラスを修正してください。 変換パターンが検証に失敗した場合は、そのパターンを修正して（reference/troubleshooting.md に記録して）から再適用してください。1つのユニットだけに個別のパッチを当ててはいけません。
5. API を作り上げないでください。 リファレンスに記載されているのは検証済みの名称のみです。記載のないものは、使用前に公式ドキュメントで確認してください。

ワークフロー

フェーズ 0：事前確認

対象 Runtime のバージョン、astro CLI の有無、リポジトリへのアクセスを確認します。プロジェクトの構成を検出します。classic（@repository／workspace.yaml）、modern（Definitions）、または Components（pyproject.toml [tool.dg]、defs.yaml ファイル）のいずれかです。これら3つはすべて実際に使われており、併用される場合もあります。この時点で移行元プロジェクトのテストスイートをベースラインとして実行します。移行前からある失敗を記録し、移行の責任には含めません。

フェーズ 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 を割り当て、マニフェストに記入してください。スキャナーは決定論的な網羅性を担い、エージェントは分類（判断）を担います。対応する mapping.md の行が見つからないレコード自体が調査結果となるため、記録し、推測しないでください。また、DAGSTER_CLOUD_ と EnvVar( も grep してください（プラットフォーム層、フェーズ 5）。

マニフェストの規則：正規のマニフェストは移行実行ディレクトリに置きます。Astro プロジェクトが作成されたら（フェーズ 2 のスキャフォールド）、マニフェストを include/inventory/manifest.json にコピーしてください。これにより、Gate 3 の pytest と status.py のデフォルトでマニフェストが見つかります。それまでは実行ディレクトリに置いたままにします（以降は両方を同期し、実行ディレクトリ内のコピーを優先します）。静的レコードが正規の移行単位です。ランタイムモードの拡充情報はこれらに統合し、真にランタイム専用の定義（ファクトリーによって生成されるもの）のみを新しい単位とします。

今すぐ移行レポートのひな形を出力してください。マニフェストの各レコードにつき 1 つのセクションと、reference/astro-deployment.md のシークレット／環境変数の命名マップを含めます。プロジェクトの規模に合わせてひな形を調整してください。シークレットのないローカルプロジェクトでは、空の定型セクションではなく、「シークレット／プラットフォーム層なし」という 1 行の注記を記載します。

フェーズ 1.5：進めるか見送るか（率直な判断基準）

何かを翻訳する前に、インベントリから答えられるプロジェクト全体の問いに答えてください。移行によってこのチームが失うものは何か、また、それぞれの損失に対して許容できる対応策があるか？ NONE と REDESIGN の行について、このチームにとって何が不可欠かを踏まえ、損失だけでなく緩和策も評価してください。

不可欠な場合	Airflow 側での対応策	次の場合に限り、現状維持のシグナルとなる
コード変更時の dbt 再構築（code_version_changed()）	cron での状態認識型 dbt ビルド（Fusion / dbt State は実行ごとに変更のないモデルをスキップするため、デプロイ後の実行で変更点だけを正確に再構築します）、および／またはマージ時に CI から起動する dbt build（PR でゲートされ、しばしばアップグレードになります）	チームが状態認識型 dbt スタックも CI からの dbt 実行もできない
マテリアライゼーションを駆動する鮮度	Astro Observe の鮮度 SLA / 適時性アラートと、SLA に合わせたスケジュール実行	鮮度トリガー型のコンピューティングが、スケジュール設定とアラートでは本当に代替できない
アセット単位のコスト計上（Insights）	Astro Observe によるパイプラインレベルのウェアハウスコスト管理。アセット単位の詳細度は失われます	アセット単位のチャージバックが契約上または組織上の要件である
日常的に使うアセットカタログ / カラムレベルのリネージ	Airflow 3 のアセットビューと OpenLineage/Astro のリネージ（アセットレベル）	外部カタログがなく、カラムレベルのリネージが日常のワークフローに組み込まれている
深くネストされた AutomationCondition の構成、can_subset、パーティション単位の選択的マテリアライゼーション	大半は cron / アセットスケジュールに分解できます（reference/automation.md を参照）。残るものはドメインごとに再設計します	分解しても何にもならない構成に、複数のドメインが依存している
センサーのカーソルのトランザクション性、実行スコープのティアダウン	冪等なコンシューマーと max_active_runs、タスク本体でのコンテキストマネージャー	冪等性では吸収できない、厳密に 1 回だけのイベント集約が正確性の要件である

この表が示すルールを明言します。dbt だけを理由に「現状維持」にはなりません。 Cosmos、状態認識型 dbt ビルド、CI からのビルドを活用すれば、dbt ワークフローの損失にはすべて、許容された実践的な緩和策があります（このスキルの評価プログラムでは、実際のウェアハウスでも実行を検証済みです）。dbt 項目は記録する条件であって、阻止要因ではありません。アセット単位のコスト計上やカラムレベルのリネージに関する可観測性の項目は別の条件であり、dbt を多用するチームであっても、それぞれ独立して評価します。

判断結果は 3 つに分かれ、最も一般的なのは中間の選択肢です。

• 移行する：現状維持のシグナルはありません。フェーズ 2 に進みます。
• 条件付きで移行する（実際のプロジェクトの大半）：損失はありますが、緩和策が特定され、レポートの第 1 セクションに書面で承認されています。特定のドメインには 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 アセットを 1 つ、パーティション分割されたアセットを 1 つ、JUDG ケースを 1 つ、または利用可能な範囲で最も近い組み合わせを選びます（小規模なプロジェクトにはパーティション分割されたアセットや MECH アセットがない場合があります。その場合は、実在する DAG を 1 つ選び、全工程を通して移行してください）。試行から得た知見は、規模を拡大する前に 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 を作成しない形で意図的に変換される単位（タスクに吸収されるヘルパー、アラートになったポリシー、接続になったリソース）は、移行先を示す証拠を記載し、target: "none" として complete にします。Gate 3 では仕様どおりスキップされます。
• 単位ごとに、まとめてコミットします。

フェーズ 5：プラットフォーム層

reference/astro-deployment.md：デプロイメントのトポロジ、シークレット／接続の命名対応表、CI/CD とプレビュー用デプロイメント、アラートポリシーのマッピング、Observe／リネージに関する期待事項、コード内の DAGSTER_CLOUD_* を書き換えるチェックリスト。

フェーズ 6：並行稼働と切り替え

Dagster を引き続き正とします。移行済みの DAG はシャドー実行または一時停止し、同じ論理期間の出力を比較します（行数とチェックサム。記録済みのマテリアライゼーションメタデータは補助的にのみ使用し、期待値は Dagster が生成した出力自体から再計算します。reference/validation.md の Gate 5 を参照）。変更ウィンドウごとに 1 つのドメインのスケジュールを切り替えます。Dagster のスケジュールを一時停止し、Airflow の DAG の一時停止を解除します。ロールバック時はこの逆を行います。切り替え後も Dagster を参照できる状態に保ちます（実行履歴は移行されません）。

Phase 7: 最終レポート

scripts/status.py summary を実行し、合格する必要があります。レポートには、Go／No-Go の評価（Phase 1.5）とその根拠、すべての定義に関する処置表、すべての同等性の行、NONE／REDESIGN による損失（系譜の深さ、code_version のトリガー、Insights のコスト計上、センサーのカーソルトランザクション性）を明記した説明、シークレットのマップ、理由を添えた延期項目の一覧を含めます。提出前に、complete の主張を10件抜き取り確認してください。

リファレンスの参照先

該当する構成要素	参照先
最初の1時間、用語集、問題が発生する可能性がある箇所	reference/quickstart.md
あらゆる項目（最初に確認する項目：構成要素ごとに1行）	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	ユニットごとの状態マシン＋完全性ゲート