環境を Cloud Composer 2 に移行する（Airflow 2 から）

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

このページでは、既存の Cloud Composer 1 の DAG、データ、構成の Airflow 2 環境を、Cloud Composer 2（Airflow 2）に移行する方法について説明します。

その他の移行ガイド

移行元	終了日	メソッド	ガイド
Cloud Composer 1、Airflow 2	Cloud Composer 2、Airflow 2	並列処理、スナップショットを使用	移行ガイド（スナップショット）
Cloud Composer 1、Airflow 1	Cloud Composer 2、Airflow 2	並列処理、スナップショットを使用	移行ガイド（スナップショット）
Cloud Composer 1、Airflow 2	Cloud Composer 2、Airflow 2	並列処理、手動転送	このガイド（手動移行）
Cloud Composer 1、Airflow 1	Cloud Composer 2、Airflow 2	並列処理、手動転送	手動移行ガイド
Airflow 1	Airflow 2	並列処理、手動転送	手動移行ガイド

始める前に

• Cloud Composer では、Cloud Composer 1 から Cloud Composer 2 への移し換え移行がサポートされています。Cloud Composer 1 から Cloud Composer 2 への置き換えアップグレードはできません。
• Cloud Composer 1 と Cloud Composer 2 の相違点の一覧を確認してください。

ステップ 1: 構成のオーバーライド、カスタム PyPI パッケージ、環境変数のリストを取得する

Console

Cloud Composer 1 環境の、構成のオーバーライド、カスタム PyPI パッケージ、環境変数のリストを取得します。

1. Google Cloud Console の [環境] ページに移動します。

   [環境] に移動

2. Cloud Composer 1 環境を選択します。

3. [環境変数] タブで、環境変数を表示します。

4. [Airflow 構成のオーバーライド] タブで、構成のオーバーライドを表示します。

5. [PyPI パッケージ] タブで、カスタム PyPI パッケージを表示します。

gcloud

環境変数のリストを取得するには、次のコマンドを実行します。

gcloud composer environments describe \
    COMPOSER_1_ENV \
    --location COMPOSER_1_LOCATION \
    --format="value(config.softwareConfig.envVariables)"

環境の Airflow 構成のオーバーライドのリストを取得するには、次のコマンドを実行します。

gcloud composer environments describe \
    COMPOSER_1_ENV \
    --location COMPOSER_1_LOCATION \
    --format="value(config.softwareConfig.airflowConfigOverrides)"

カスタム PyPI パッケージのリストを取得するには、次のコマンドを実行します。

gcloud composer environments describe \
    COMPOSER_1_ENV \
    --location COMPOSER_1_LOCATION \
    --format="value(config.softwareConfig.pypiPackages)"

次のように置き換えます。

• COMPOSER_1_ENV は、Cloud Composer 1 環境の名前。
• COMPOSER_1_LOCATION は、Cloud Composer 1 環境が配置されているリージョン。

Terraform

このステップをスキップします。Cloud Composer 1 環境の構成には、環境の構成のオーバーライド、カスタム PyPI パッケージ、環境変数がすでに記載されています。

ステップ 2: Cloud Composer 2 環境を作成する

このステップでは、Cloud Composer 2 環境を作成します。想定されるリソース需要を満たす環境プリセットから始めて、後で環境をさらにスケーリングして最適化できます。

Console

Cloud Composer 2 環境を作成し、構成のオーバーライドと環境変数を指定します。

それ以外の方法では、環境を作成した後に Airflow の構成と環境変数をオーバーライドできます。

gcloud

Cloud Composer 2 環境を作成し、構成のオーバーライドと環境変数を指定します。

それ以外の方法では、環境を作成した後に Airflow の構成と環境変数をオーバーライドできます。

Terraform

Cloud Composer 2 環境は、Cloud Composer 1 環境の構成に基づいて作成します。

1. Cloud Composer 1 環境の構成をコピーします。
2. 環境の名前を変更します。

3. google-beta プロバイダを使用します。

   resource "google_composer_environment" "example_environment_composer_2" {
     provider = google-beta
     # ...
   }
   

4. config.software_config ブロックに、Cloud Composer 2 イメージを指定します。

   software_config {
     image_version = "composer-2.8.3-airflow-2.7.3"
     # ...
   }
   

5. まだ行っていない場合は、構成のオーバーライドと環境変数を指定します。

6. config.software_config.pypi_packages ブロックに、カスタム PyPI パッケージを指定します。

   software_config {
   
     # ...
   
     pypi_packages = {
       numpy = ""
       scipy = ">=1.1.0"
     }
   
   }
   

ステップ 3: PyPI パッケージを Cloud Composer 2 環境にインストールする

注: 環境でカスタム PyPI パッケージを使用していない場合、この手順は省略してください。カスタム PyPI パッケージとは、環境にプリインストールされたパッケージに加えてインストールするパッケージです。

Cloud Composer 2 環境を作成したら、その環境にカスタム PyPI パッケージをインストールします。

Console

1. Google Cloud Console の [環境] ページに移動します。

   [環境] に移動

2. Cloud Composer 2 環境を選択します。

3. [PyPI パッケージ] タブに移動し、[編集] をクリックします。

4. Cloud Composer 1 環境から PyPI パッケージに必要なものをコピーします。[保存] をクリックして、環境が更新されるまで待ちます。

gcloud

1. カスタム PyPI パッケージのリストを含む requirements.txt ファイルを作成します。

     numpy
     scipy>=1.1.0
   

2. 環境を更新します。また、requirements.txt ファイルは、--update-pypi-packages-from-file コマンドに渡します。

   gcloud composer environments update COMPOSER_2_ENV \
     --location COMPOSER_2_LOCATION  \
     --update-pypi-packages-from-file requirements.txt
   

   次のように置き換えます。

   • COMPOSER_2_ENV は、Cloud Composer 2 環境の名前。
   • COMPOSER_2_LOCATION は、Cloud Composer 2 環境が配置されているリージョン。

Terraform

このステップをスキップします。カスタム PyPI パッケージは、環境の作成時にすでにインストールしています。

ステップ 4: 変数とプールを移行する

注: 環境で default_pool 以外の Airflow 変数やプールを使用しない場合は、この手順を省略してください。

Airflow では、変数とプールを JSON ファイルにエクスポートできます。そうすると、これらのファイルを Cloud Composer 2 環境にインポートできます。

このステップで使用する Airflow CLI コマンドは、Airflow ワーカーのローカル ファイルに対して動作します。ファイルをアップロードまたはダウンロードするには、使用している環境の Cloud Storage バケットの /data フォルダを使用します。このフォルダは Airflow ワーカーの /home/airflow/gcs/data/ ディレクトリに同期されます。Airflow CLI コマンドで、FILEPATH パラメータに /home/airflow/gcs/data/ を指定します。

gcloud

1. Cloud Composer 1 環境の変数をエクスポートします。

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
        variables export -- /home/airflow/gcs/data/variables.json
   

   次のように置き換えます。

   • COMPOSER_1_ENV は、Cloud Composer 1 環境の名前。
   • COMPOSER_1_LOCATION は、Cloud Composer 1 環境が配置されているリージョン。

2. Cloud Composer 1 環境のプールをエクスポートします。

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
        pools export -- /home/airflow/gcs/data/pools.json
   

   次のように置き換えます。

   • COMPOSER_1_ENV は、Cloud Composer 1 環境の名前。
   • COMPOSER_1_LOCATION は、Cloud Composer 1 環境が配置されているリージョン。

3. Cloud Composer 2 環境のバケット URI を取得します。

   1. 次のコマンドを実行します。

      gcloud composer environments describe COMPOSER_2_ENV \
          --location COMPOSER_2_LOCATION \
           --format="value(config.dagGcsPrefix)"
      

      次のように置き換えます。

      • COMPOSER_2_ENV は、Cloud Composer 2 環境の名前。
      • COMPOSER_2_LOCATION は、環境が配置されているリージョン。

   2. 出力で、/dags フォルダを削除します。結果は、Cloud Composer 2 環境のバケットの URI です。

      たとえば、gs://us-central1-example-916807e1-bucket/dags を gs://us-central1-example-916807e1-bucket に変更します。

4. 変数とプールを含む JSON ファイルを、Cloud Composer 2 環境に移行します。

   gcloud composer environments storage data export \
       --destination=COMPOSER_2_BUCKET/data \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION \
       --source=variables.json
   

   gcloud composer environments storage data export \
       --destination=COMPOSER_2_BUCKET/data \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION \
       --source=pools.json
   

   次のように置き換えます。

   • COMPOSER_2_BUCKET は、前の手順で取得した Cloud Composer 2 環境バケットの URI。
   • COMPOSER_1_ENV は、Cloud Composer 1 環境の名前。
   • COMPOSER_1_LOCATION は、Cloud Composer 1 環境が配置されているリージョン。

5. 変数とプールを Cloud Composer 2 にインポートします。

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       variables import \
       -- /home/airflow/gcs/data/variables.json
   

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       pools import \
       -- /home/airflow/gcs/data/pools.json
   

6. 変数とプールがインポートされたことを確認します。

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       variables list
   

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       pools list
   

7. バケットから JSON ファイルを削除します。

   gcloud composer environments storage data delete \
       variables.json \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION
   

   gcloud composer environments storage data delete \
       pools.json \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION
   

   gcloud composer environments storage data delete \
       variables.json \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION
   

   gcloud composer environments storage data delete \
       pools.json \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION
   

ステップ 5: Cloud Composer 1 環境のバケットから他のデータを移行する

Cloud Composer 1 環境のバケットからプラグインや他のデータを移行します。

gcloud

1. プラグインを Cloud Composer 2 環境に移行します。これを行うには、Cloud Composer 1 環境のバケットから Cloud Composer 2 環境のバケットにある /plugins フォルダに、プラグインをエクスポートします。

   gcloud composer environments storage plugins export \
       --destination=COMPOSER_2_BUCKET/plugins \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION
   

2. /plugins フォルダが正常にインポートされたことを確認します。

   gcloud composer environments storage plugins list \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION
   

3. /data フォルダを、Cloud Composer 1 環境から Airflow 2 環境にエクスポートします。

   gcloud composer environments storage data export \
       --destination=COMPOSER_2_BUCKET/data \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION
   

4. /data フォルダが正常にインポートされたことを確認します。

   gcloud composer environments storage data list \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION
   

ステップ 6: 接続を移行する

注意: Cloud Composer 環境には、環境用にあらかじめ構成されたデフォルトの接続が多数あります。Cloud Composer 2 環境には、これらの接続がすでに存在するため、これらの接続は移行しないでください。環境でデフォルトの接続のみを使用している場合は、この手順を省略します。

このステップでは、手動で接続を作成して移行する方法について説明します。

gcloud

1. Cloud Composer 1 環境の接続のリストを取得するために、次のコマンドを実行します。

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
       connections list
   

2. Cloud Composer 2 環境で新しい接続を作成するために、gcloud を使用して connections Airflow CLI コマンドを実行します。次に例を示します。

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       connections add \
       -- --conn-host postgres.example.com \
       --conn-port 5432 \
       --conn-type postgres \
       --conn-login example_user \
       --conn-password example_password \
       --conn-description "Example connection" \
       example_connection
   

ステップ 7: ユーザー アカウントを移行する

このステップでは、手動で接続を作成してユーザーを移行する方法について説明します。

注: ユーザーを表示するには、アカウントに Admin ロールが必要です。

Airflow UI

1. Cloud Composer 1 環境でユーザーの一覧を表示するには、次のようにします。

   1. Cloud Composer 1 環境の Airflow ウェブ インターフェースを開きます。

   2. [セキュリティ] > [ユーザーを一覧表示] に移動します。

2. Cloud Composer 2 環境でユーザーを作成するには、次のようにします。

   1. Cloud Composer 2 環境の Airflow ウェブ インターフェースを開きます。

   2. [セキュリティ] > [ユーザーを一覧表示] に移動します。

   3. [新しいレコードの追加] をクリックします。

gcloud

1. Cloud Composer 1 環境でユーザーのリストを表示するには、gcloud で users list Airflow CLI コマンドを実行します。

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
       users list
   

   次のように置き換えます。

   • COMPOSER_1_ENV は、Cloud Composer 1 環境の名前。
   • COMPOSER_1_LOCATION は、Cloud Composer 1 環境が配置されているリージョン。

2. Cloud Composer 2 環境で新しいユーザー アカウントを作成するには、gcloud で users create Airflow CLI コマンドを実行します。次に例を示します。

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       users create \
       -- --username example_username \
       --firstname Example-Name \
       --lastname Example-Surname \
       --email [e-mail] \
       --use-random-password \
       --role Op
   

   次のように置き換えます。

   • COMPOSER_2_ENV は、Cloud Composer 2 環境の名前。
   • COMPOSER_2_LOCATION は、Cloud Composer 2 環境が配置されているリージョン。
   • ユーザーのロールを含む、Cloud Composer 1 環境のすべてのユーザー構成パラメータとその値。

別の方法でユーザー アカウントを移行する

別の方法としては、users export と users import の Airflow CLI コマンドを使用できます。

1. 環境のバケット /data フォルダにあるファイルにユーザー アカウントをエクスポートします。

   gcloud composer environments run \
       COMPOSER_1_ENV \
       --location COMPOSER_1_LOCATION \
       users export -- /home/airflow/gcs/data/users.json
   

2. このファイルを Cloud Composer 2 環境のバケットにエクスポートします。

   gcloud composer environments storage data export \
       --destination=COMPOSER_2_BUCKET/data \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION \
       --source=users.json
   

3. このファイルから Cloud Composer 2 環境にユーザー アカウントをインポートします。

   gcloud composer environments run \
       COMPOSER_2_ENV \
       --location COMPOSER_2_LOCATION \
       users import \
       -- /home/airflow/gcs/data/users.json
   

4. 両方の環境で、その JSON ファイルを削除します。

   gcloud composer environments storage data delete \
       --environment=COMPOSER_1_ENV \
       --location=COMPOSER_1_LOCATION  \
       users.json
   

   gcloud composer environments storage data delete \
       --environment=COMPOSER_2_ENV \
       --location=COMPOSER_2_LOCATION  \
       users.json
   

次のように置き換えます。

• COMPOSER_1_ENV は、Cloud Composer 1 環境の名前。
• COMPOSER_2_ENV は、Cloud Composer 2 環境の名前。
• COMPOSER_1_LOCATION は、Cloud Composer 1 環境が配置されているリージョン。
• COMPOSER_2_LOCATION は、Cloud Composer 2 環境が配置されているリージョン。
• COMPOSER_2_BUCKET は、前の手順で取得した Cloud Composer 2 環境バケットの URI。

ステップ 8: DAG を Cloud Composer 2 環境に移行する

警告: このステップでは、環境間で DAG を移行するときの重複、欠落、エラー、無効な DAG 実行に関する潜在的な問題を回避する方法について説明します。

DAG を環境間で転送すると、次の問題が発生する場合があります。

• 両方の環境で DAG が有効である（一時停止されていない）場合、各環境はスケジュールされたとおりに DAG の独自のコピーを実行します。これにより、同じデータと実行時間の重複した DAG 実行が発生する可能性があります。

• DAG のキャッチアップのために、Airflow は DAG で指定された開始日に始まる追加の DAG 実行をスケジュールします。これは、新しい Airflow インスタンスで Cloud Composer 1 環境からの DAG 実行の履歴が考慮されないためです。このため、指定した開始日に始まるように大量の DAG 実行がスケジュールされる可能性があります。

DAG の重複実行を防止する

Cloud Composer 2 環境の Airflow 2 環境で、dags_are_paused_at_creation オプションに対する Airflow 構成オプションのオーバーライドを追加します。この変更を行うと、デフォルトで新しいすべての DAG が一時停止されます。

セクション	キー	値
core	dags_are_paused_at_creation	True

注: このオーバーライドは、すべての DAG を Cloud Composer 2 環境に移行した後に削除できます。

DAG 実行の追加や欠落を防ぐ

実行日のギャップや重複を回避するには、Cloud Composer 2 でのキャッチアップを無効にします。このようにして、DAG を Cloud Composer 2 環境にアップロードすると、Airflow では Cloud Composer 1 環境ですでに実行されている DAG 実行がスケジュールされません。catchup_by_default オプションに対する Airflow 構成オプションのオーバーライドを追加します。

セクション	キー	値
scheduler	catchup_by_default	False

注意: このオーバーライドは削除しないでください。削除すると、環境により DAG 実行が補われます。これらの DAG 実行は、Cloud Composer 1 環境ですでに実行されているものと重複する場合があります。

DAG を Cloud Composer 2 環境に移行する

DAG を Cloud Composer 2 環境に移行するには、次のようにします。

1. Cloud Composer 1 環境から Cloud Composer 2 環境に DAG をアップロードします。airflow_monitoring.py DAG をスキップします。

2. 構成のオーバーライドにより、DAG は Cloud Composer 2 環境で一時停止され、DAG 実行はスケジュール設定されません。

3. Airflow ウェブ インターフェースで、[DAG] に移動し、報告された DAG 構文エラーを確認します。

4. DAG の転送を計画する時刻:

   1. Cloud Composer 1 環境で、DAG を一時停止します。

   2. Cloud Composer 2 環境で、DAG の一時停止を解除します。

   3. 新しい DAG 実行が正しい時間にスケジュール設定されていることを確認します。

   4. Cloud Composer 2 環境で DAG の実行が発生するのを待ち、成功したかどうかを確認します。DAG の実行が成功した場合は、Cloud Composer 1 環境で一時停止を解除しないでください。解除すると、DAG が Cloud Composer 1 の環境で同じ日時に実行されます。

5. 特定の DAG の実行が失敗した場合は、Cloud Composer 2 で正常に実行されるまで DAG のトラブルシューティングを行います。

   必要な場合は、次のようにいつでも DAG の Cloud Composer 1 バージョンにフォールバックして、Cloud Composer 2 で失敗した DAG 実行を Cloud Composer 1 環境で実行できます。

   1. Cloud Composer 2 環境で DAG を一時停止します。

   2. Cloud Composer 1 環境で DAG の一時停止を解除します。これにより、Cloud Composer 1 環境で、一時停止された時間の DAG 実行を補う DAG がスケジュールされます。

ステップ 9: Cloud Composer 2 環境をモニタリングする

すべての DAG と構成を Cloud Composer 2 環境に移行した後、潜在的な問題、失敗した DAG 実行、環境全体の健全性をモニタリングします。Cloud Composer 2 環境が、十分な時間、問題なく動作している場合は、Cloud Composer 1 環境の削除を検討してください。

次のステップ

• DAG のトラブルシューティング
• 環境作成のトラブルシューティング
• 環境更新のトラブルシューティング
• バックポート パッケージの使用