managed-airflow-migrations

Managed Service for Apache Airflow（旧 Cloud Composer）移行ガイド

このスキルでは、既存の Managed Service for Apache Airflow（旧 Cloud Composer）環境（またはローカルで利用可能な環境）の Airflow DAG を調整し、Airflow 2.11.1（MSAA Gen 2 または 3）または Airflow 3（MSAA Gen 3）と互換性を持たせる手順を説明します。

---

帰属情報

必ずお読みください：gcloud、REST API、または MCP コマンドを実行する前に、必ず利用状況の帰属情報をお読みください。ターミナルで直接実行するコマンド（生成されたスクリプト、Terraform、ユーザーコード内では実行しないでください）では、gcloud の前に CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:managed-airflow-migrations)" を付け、User-Agent ヘッダーに agent-skills/1.0 (skill:managed-airflow-migrations) を追加してください。

フェーズ 1：調査とダウンロード

変更を加える前に、明示的に依頼された場合は既存の DAG ファイルをダウンロードしてください。明示的に依頼された場合に限り、ソース環境を調査してソースのバージョンを確認してください。環境の調査とファイルのダウンロードに関する詳しい手順は、 references/environment-inspection.mdを確認してください。

---

フェーズ 2：ターゲットバージョンと依存関係のマッピング

2.1 Airflow 2.11.1 以降の依存関係のマッピング

Airflow 2.11.1（MSAA Gen 2）または Airflow 3 に移行する場合は、以下の一覧を使用して、主要な依存関係のバージョンの推移を確認してください。この一覧では、Airflow 2.11.1 に到達するために必要な変更を説明しています。Airflow 2（2.11.1 より前）から Airflow 3 に移行する場合は、これらの変更も考慮してください。

Composer 2.10.0（Airflow 2.10.2）

• Google Provider：10.26.0
• SSH Provider：3.14.0
• HTTP Provider：4.13.3
• 破壊的変更：完全に文書化されている最も古いソースの基準です。

Composer 2.15.3（Airflow 2.10.5）

• Google Provider：18.0.0
• SSH Provider：4.1.4
• HTTP Provider：5.3.4
• 破壊的変更：
  • SSH Provider 4.0.0： Hook の timeout が削除されました。get_conn() はコンテキストマネージャーになりました。
  • HTTP Provider 5.0.0： SimpleHttpOperator -> HttpOperator。
  • Google Provider 11.0.0： BigQueryExecuteQueryOperator が削除されました。
  • Google Provider 12.0.0： 従来の Data Pipeline オペレーターが削除されました。
  • Google Provider 13.0.0： AutoMLBatchPredictOperator が削除されました。
  • Google Provider 17.0.0： BigQueryCreateEmptyTableOperator と BigQueryCreateExternalTableOperator が削除されました。Life Sciences オペレーターも削除されました。
  • Google Provider 18.0.0： 従来の DV360 オペレーターが削除されました。

Composer 2.16.1（Airflow 2.10.5）

• Google Provider：19.0.0
• SSH Provider：4.1.6
• HTTP Provider：5.5.0
• 破壊的変更：Google Provider 19.0.0： AutoML オペレーターが削除されました（Vertex AI を使用してください）。

Composer 2.17.0（ターゲット Airflow 2.11.1）

• Google Provider：20.0.0
• SSH Provider：5.0.0
• HTTP Provider：6.0.2
• 破壊的変更：
  • SSH Provider 5.0.0： sshtunnel が削除されました（ネイティブトンネリングに変更）。
  • HTTP Provider 6.0.0： JSON シリアル化。
  • Google Provider 20.0.0： ADLS Gen2 への移行。

2.2 Airflow 3 への移行

Airflow 3（MSAA Gen 3）に移行する場合は、以下を含む大幅な変更を伴うメジャーバージョンアップグレードであることに注意してください。

• Task SDK の分離（インポート元が airflow から airflow.sdk に変更）。
• メタデータ DB への直接アクセスの削除。
• Dataset の Asset への名称変更。
• SubDAG と SLA の削除。
• コンテキスト変数の利用可否の変更。

Airflow 2 に適用されるすべての変更を考慮してください（たとえば、Airflow 2.10.2 から移行する場合は、Airflow 2.11.1 への移行に必要な変更を適用したうえで、Airflow 3 への移行に関する変更も適用してください）。

---

フェーズ 3：分析と修正（ダウンロードしたファイルのスキャン）

スキャンコマンドは、ローカルワークスペースのルートから実行します （特に指定がない限り、./migration_workspace）。

---

3.1 Airflow 2.11.1 のコアと依存関係のチェック

Airflow 2.11.1 以降への移行時（Airflow 3 への移行における中間ステップ）に、これらのスキャンを使用します。

3.1.1 データセットによるスケジューリング（Airflow 2.11.0）

• 変更点： データセットでスケジュールされた DAG は、DAG が一時停止されていない間にイベントが発生した場合にのみ実行されます。
• スキャンコマンド： grep -rn "Dataset(" ./dags
• 修正方法： これらの DAG はイベントを検知できるように一時停止せずにおく必要があることを記録するか、追いつくための手動トリガーを計画する必要があります。

3.1.2 説明内の HTML（Airflow 2.11.0）

• 変更点： DAG のドキュメント／パラメーター内の生の HTML は、デフォルトでエスケープされます。

• スキャンコマンド：

  grep -rn -E "doc_md.*<|doc_md.*>|description.*<|description.*>" ./dags
  

• 修正方法： HTML を Markdown に変換するか、移行先で AIRFLOW__WEBSERVER__ALLOW_RAW_HTML_DESCRIPTIONS=True を設定します。

3.1.3 ティアダウンタスク（Airflow 2.10.5）

• 変更点： DAG が失敗としてマークされると、ティアダウンタスクは常に実行されます。
• スキャンコマンド： grep -rn "as_teardown" ./dags
• 修正方法： ティアダウンタスクが冪等であることを確認します。

3.1.4 Pendulum 3 へのアップグレード（Airflow 2.11.0）

• 変更点： Period は Interval に名称変更され、テスト用ヘルパーは削除されました。

• スキャンコマンド（コード）：

  grep -rn -E "pendulum\.Period|pendulum\.period" ./dags
  

• スキャンコマンド（テスト）：

  grep -rn -E "\.test\(|set_test_now\(" ./tests 2>/dev/null || true
  

• 修正方法： Period を Interval に、period(...) を interval(...) に置き換えます。

---

3.2 パス A：Airflow 2.11.1 プロバイダーパッケージのスキャン

3.2.1 SSH プロバイダー（SSH 4.0.0 および 5.0.0）

• スキャンコマンド（タイムアウト）： grep -rn "SSHHook" ./dags | grep "timeout"
• スキャンコマンド（コンテキストマネージャー）： grep -rn "with SSHHook" ./dags
• スキャンコマンド（トンネル属性）： grep -rn "\.get_tunnel" ./dags
• 修正方法：
  • SSHHook の timeout を conn_timeout に置き換えます。
  • with hook as conn: を with hook.get_conn() as conn: に置き換えます。
  • get_tunnel() をコンテキストマネージャーとして使用します：with hook.get_tunnel(...) as tunnel:。

3.2.2 HTTP プロバイダー（HTTP 5.0.0 および 6.0.0）

• スキャンコマンド： grep -rn "SimpleHttpOperator" ./dags
• 修正方法： SimpleHttpOperator を HttpOperator に置き換えます。

3.2.3 Google プロバイダー（v11 から v20）

• スキャンコマンド（BigQuery クエリ）：

  grep -rn "BigQueryExecuteQueryOperator" ./dags
  

  • 修正方法： BigQueryInsertJobOperator に置き換えます（ configuration 辞書を使用します）。

• スキャンコマンド（BigQuery テーブル）：

  grep -rn -E "BigQueryCreateEmptyTableOperator|BigQueryCreateExternalTableOperator" ./dags
  

  • 修正方法： BigQueryCreateTableOperator に置き換えます（ table_resource 辞書を使用します）。

• スキャンコマンド（AutoML）：

  grep -rn -E "AutoMLTrainModelOperator|AutoMLPredictOperator|AutoMLCreateDatasetOperator|AutoMLBatchPredictOperator" ./dags
  

  • 修正方法： Vertex AI オペレーターに移行します。

• スキャンコマンド（Dataflow）：

  grep -rn -E "CreateDataPipelineOperator|RunDataPipelineOperator" ./dags
  

  • 修正方法： DataflowCreatePipelineOperator／DataflowRunPipelineOperator に置き換えます。

• スキャンコマンド（Life Sciences）：

  grep -rn "LifeSciencesRunPipelineOperator" ./dags`
  

  • 修正方法： Google Cloud Batch オペレーター（ BatchCreateJobOperator）に移行します。

• スキャンコマンド（ADLS から GCS）： grep -rn "ADLSToGCSOperator" ./dags

  • 修正方法： file_system_name が指定されていることを確認します。

---

3.3 Airflow 3 の移行チェック

Airflow 3 に移行する際は、references/airflow-3.md の手順を使用します。

---

フェーズ 4：デプロイと検証

デプロイと検証の手順は、明示的に依頼された場合にのみ実施してください。

4.1 静的検証（Airflow 3 に移行する場合）

Airflow 3 用のコード変更を適用した後、構文が正しいことを確認します。開発環境で利用できる場合は、静的リントチェックを実行します。

ruff check {target_dag_file} --select AIR30

報告された非推奨警告は、完了前にすべて解消してください。ruff が利用できない場合は、インストールを推奨してください。

4.2 MSAA へのデプロイ

4.2.1 対象 GCS バケットのパスを取得する（依頼された場合のみ）

gcloud composer environments describe <TARGET_ENV> \
    --location <TARGET_REGION> \
    --format="value(config.dagGcsPrefix)"

想定される出力： gs://<target-bucket-name>/dags

4.2 変更した DAG とバケットの依存関係をアップロードする（依頼された場合のみ）

この手順は、明示的に依頼された場合にのみ実施してください。 変更した DAG と、バックアップしたバケットの依存関係を、ローカルのワークスペースから対象の GCS バケットにコピーします。確認手順を省略した場合は、正しい <target-bucket-name> を使用していることを確認してください。

1. DAG をアップロードする：

   gcloud storage cp -r ./dags/* gs://<target-bucket-name>/dags/
   

2. その他のバケットの依存関係をアップロードする（該当する場合）：

   gcloud storage cp -r ./migration_workspace/<dependency-folder> gs://<target-bucket-name>/<dependency-folder>
   

4.3 Airflow CLI で DAG を検証する

この手順は、変更した DAGS をターゲット環境にアップロードするよう明示的に依頼された場合にのみ実施します（アップロード後に実施してください）。

Airflow CLI を使用して、DAG がターゲット環境に正常にアップロードされ、解析され、Airflow スケジューラーに登録されたことを確認できます。

1. 登録済みの DAG を一覧表示する： 次のコマンドを実行して、ターゲット環境に登録されているすべての DAG を一覧表示します。移行した DAG がこの一覧に含まれていることを確認してください。

   gcloud composer environments run <TARGET_ENV> \
       --location <TARGET_REGION> \
       dags list
   

2. インポートエラーを確認する： 一覧に表示されない DAG がある場合、または解析上の問題がないことを確認する場合は、インポートエラーを確認します。

   gcloud composer environments run <TARGET_ENV> \
       --location <TARGET_REGION> \
       dags list-import-errors
   

   想定される出力：

   • エラーがない場合、コマンドは No data found を出力します。
   • エラーがある場合は、ファイルパスとエラーのトレースバックが一覧表示されます。

注：Airflow スケジューラーが新しいファイルを解析し、変更がこれらのコマンドの結果に反映されるまで、数分かかることがあります。

4.4 Cloud Logging で検証する

この手順は、変更した DAGS をターゲット環境にアップロードするよう明示的に依頼された場合にのみ実施します（アップロード後に実施してください）。 ターゲット環境の Cloud Logging を監視して、実行時エラーやインポートエラーを検出します。

GCP Cloud Logging Console で次のクエリを実行します（または gcloud logging read を使用します）。

resource.type="cloud_composer_environment"
resource.labels.environment_name="<TARGET_ENV>"
log_id("airflow-scheduler")
severity>=ERROR

---

付録：ローカル環境での検証

ターゲット環境にデプロイする前にローカルで変更を検証する場合は、Composer Local Development CLI tool （composer-dev）を使用できます。ローカル開発環境とのやり取りについては、 references/local-development-environment.md を参考にしてください。