データ処理ワークフロー用の CI / CD パイプラインの設定

Last reviewed 2022-08-15 UTC

このチュートリアルでは、Google Cloud のマネージドプロダクトで継続的インテグレーション / 継続的デプロイ（CI / CD）メソッドを実装することで、データ処理のための CI / CD パイプラインを設定する方法について説明します。データサイエンティストやアナリストは CI / CD の方法論を応用して、品質、保守性、適応性に優れたデータ処理とワークフローを実現できます。適用できる手法は次のとおりです。

ソースコードのバージョン管理
アプリの自動的なビルド、テスト、デプロイ
本番環境からの環境分離
環境設定のための複製可能な手順

このチュートリアルは、繰り返し実行されるデータ処理ジョブを構築するデータサイエンティストやアナリストを対象としており、研究開発（R&D）を構造化してデータ処理ワークロードを体系的かつ自動的に維持するのに役立ちます。

デプロイアーキテクチャ

このガイドでは、次の Google Cloud プロダクトを使用します。

Cloud Build: データ処理ワークフローとデータ処理自体をビルド、デプロイ、テストする CI/CD パイプラインを作成します。Cloud Build は、Google Cloud 上でビルドを実行するマネージドサービスです。ビルドは、各ステップが Docker コンテナで実行される一連のビルドステップです。
Cloud Composer: データ処理の開始、テスト、結果の検証など、ワークフローのステップを定義して実行します。Cloud Composer は、マネージド Apache Airflow サービスです。このサービスは、このチュートリアルのデータ処理ワークフローのような複雑なワークフローを作成、スケジュール設定、モニタリング、管理できる環境を提供します。
Dataflow: サンプルデータ処理として Apache Beam WordCount の例を実行します。

CI / CD パイプライン

大まかに言えば、CI/CD パイプラインは次のステップで構成されます。

Cloud Build が Maven ビルダーを使用して WordCount サンプルを自己実行型の Java アーカイブ（JAR）ファイルにパッケージ化します。Maven ビルダーは、Maven がインストールされているコンテナです。Maven ビルダーを使用するようにビルドステップが構成されている場合、Maven がタスクを実行します。
Cloud Build が JAR ファイルを Cloud Storage にアップロードします。
Cloud Build がデータ処理ワークフローコードの単体テストを実行し、そのワークフローコードを Cloud Composer にデプロイします。
Cloud Composer が JAR ファイルを受け取り、Dataflow 上でデータ処理ジョブを実行します。

次の図に、CI / CD パイプラインステップの詳細を示します。

CI / CD パイプラインのアーキテクチャ図。

このチュートリアルでは、テスト環境と本番環境へのデプロイが 2 つの異なる Cloud Build パイプライン（テストパイプラインと本番環境パイプライン）に分かれています。

上の図では、テストパイプラインは以下のステップで構成されています。

デベロッパーがコードの変更を Cloud Source Repositories に commit します。
コードの変更によって Cloud Build でテストビルドがトリガーされます。
Cloud Build が自己実行型 JAR ファイルをビルドし、それを Cloud Storage のテスト用 JAR バケットにデプロイします。
Cloud Build がテストファイルを Cloud Storage のテストファイルバケットにデプロイします。
Cloud Build が、新しくデプロイされた JAR ファイルを参照するように Cloud Composer の変数を設定します。
Cloud Build が、データ処理ワークフローの有向非巡回グラフ（DAG）をテストし、Cloud Storage の Cloud Composer バケットにデプロイします。
このワークフローの DAG ファイルが Cloud Composer にデプロイされます。
Cloud Build が、新しくデプロイされたデータ処理ワークフローの実行をトリガーします。
データ処理ワークフロー統合テストに合格すると、メッセージは Pub/Sub に公開され、そこにはメッセージのデータフィールド内の最新の自己実行型 JAR（Airflow 変数から取得）への参照が含まれます。

上の図では、本番環境パイプラインは以下のステップで構成されています。

メッセージが Pub/Sub トピックに公開されると、本番環境デプロイパイプラインがトリガーされます。
デベロッパーが本番環境デプロイパイプラインを手動で承認し、ビルドが実行されます。
Cloud Build が、最新の自己実行型 JAR ファイルを Cloud Storage のテスト用 JAR バケットから本番環境用 JAR バケットにコピーします。
Cloud Build が、本番環境用データ処理ワークフロー DAG をテストし、Cloud Storage の Cloud Composer バケットにデプロイします。
本番環境用のワークフロー DAG ファイルが Cloud Composer にデプロイされます。

このチュートリアルでは、本番環境用データ処理ワークフローをテスト用ワークフローと同じ Cloud Composer 環境にデプロイして、すべてのデータ処理ワークフローを一元的に確認できるようにします。このチュートリアルでは、異なる Cloud Storage バケットを使用して入力データと出力データを保持し、環境を分離します。

環境を完全に分離するには、異なるプロジェクト内に作成された複数の Cloud Composer 環境が必要です。このようにして作成された環境はデフォルトで分離され、本番環境を保護するのに役立ちます。ただし、このアプローチは本チュートリアルの対象範囲外です。複数の Google Cloud プロジェクトのリソースにアクセスする方法の詳細については、サービスアカウント権限の設定をご覧ください。

データ処理ワークフロー

Cloud Composer がデータ処理ワークフローを実行する手順は、Python で書かれた有向非巡回グラフ（DAG）で定義されます。DAG では、データ処理ワークフローのすべてのステップが、それぞれの依存関係とともに定義されます。

CI / CD パイプラインは毎回のビルドの中で、DAG 定義を Cloud Source Repositories から Cloud Composer に自動的にデプロイします。このプロセスにより、Cloud Composer のワークフロー定義は人の手を介さなくても常に最新の状態に保たれます。

テスト環境用の DAG 定義では、データ処理ワークフローに加えてエンドツーエンドのテストステップが定義されています。テストステップは、データ処理ワークフローが正しく実行されることを確認するのに役立ちます。

次の図に、データ処理ワークフローを示します。

4 ステップのデータ処理ワークフロー。

データ処理ワークフローは次のステップで構成されます。

Dataflow で WordCount データ処理を実行します。
WordCount プロセスの出力ファイルをダウンロードします。WordCount プロセスは次の 3 つのファイルを出力します。
- download_result_1
- download_result_2
- download_result_3
download_ref_string という名前の参照ファイルをダウンロードします。
この参照ファイルと照らし合わせて結果を検証します。この統合テストでは、3 つの結果すべてを集計して、集計結果を参照ファイルと比較します。
統合テストに合格したら、メッセージを Pub/Sub に公開します。

Cloud Composer などのタスクオーケストレーションフレームワークを使用してデータ処理ワークフローを管理すると、ワークフローのコードの複雑さを軽減できます。

テスト

このチュートリアルには、データ処理ワークフローをエンドツーエンドで検証する統合テストのほかに、2 つの単体テストがあります。それは、データ処理コードとデータ処理ワークフローコードの自動テストです。データ処理コードのテストは Java で記述されていて、Maven ビルドプロセス中に自動的に実行されます。データ処理ワークフローコードのテストは Python で記述されていて、独立したビルドステップとして実行されます。

目標

Cloud Composer 環境を構成する。
データ用の Cloud Storage バケットを作成する。
ビルド、テスト、本番環境用のパイプラインを作成する。
ビルドトリガーを構成する。

費用

このドキュメントでは、Google Cloud の次の課金対象のコンポーネントを使用します。

料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを生成できます。新しい Google Cloud ユーザーは無料トライアルをご利用いただける場合があります。

このドキュメントに記載されているタスクの完了後、作成したリソースを削除すると、それ以上の請求は発生しません。詳細については、クリーンアップをご覧ください。

始める前に

Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

Google Cloud Console の [プロジェクトセレクタ] ページで、Google Cloud プロジェクトを選択または作成します。

プロジェクトセレクタに移動

Google Cloud プロジェクトで課金が有効になっていることを確認します。

Cloud Build, Cloud Source Repositories, Cloud Composer, and Dataflow API を有効にします。

API を有効にする

Google Cloud Console の [プロジェクトセレクタ] ページで、Google Cloud プロジェクトを選択または作成します。

プロジェクトセレクタに移動

Google Cloud プロジェクトで課金が有効になっていることを確認します。

Cloud Build, Cloud Source Repositories, Cloud Composer, and Dataflow API を有効にします。

API を有効にする

サンプルコード

サンプルコードは次の 2 つのフォルダにあります。

env-setup フォルダには、Google Cloud 環境の初期設定用のシェルスクリプトが含まれています。
source-code フォルダには、時間の経過に伴って継続的に開発され、ソース管理が必要なコードが含まれています。このコードによってビルドとテストの自動的なプロセスがトリガーされます。このフォルダには次のサブフォルダが含まれます。
- data-processing-code フォルダには、Apache Beam プロセスのソースコードが含まれています。
- workflow-dag フォルダには、データ処理ワークフローの Composer DAG 定義が含まれています。この DAG 定義には、Dataflow プロセスを設計、実装、テストするステップが記述されています。
- build-pipeline フォルダには 2 つの Cloud Build 構成が含まれています。1 つはテストパイプライン用の構成であり、もう 1 つは本番環境パイプライン用の構成です。このフォルダには、これらのパイプラインのサポートスクリプトも含まれています。

このチュートリアルの目的上、データ処理用と DAG ワークフロー用のソースコードファイルは、同じソースコードリポジトリ内の別のフォルダに格納されています。本番環境では、通常これらのソースコードファイルは個別のソースコードリポジトリに格納され、別のチームによって管理されます。

環境設定

このチュートリアルでは、すべてのコマンドを Cloud Shell で実行します。Cloud Shell は、Google Cloud コンソールの下部にウィンドウとして表示されます。

Google Cloud Console で Cloud Shell を開きます。

Cloud Shell を開く

サンプルコードリポジトリのクローンを作成します。

git clone https://github.com/GoogleCloudPlatform/ci-cd-for-data-processing-workflow.git

スクリプトを実行して環境変数を設定します。
```
cd ~/ci-cd-for-data-processing-workflow/env-setup
source set_env.sh
```
このスクリプトでは次の環境変数を設定します。
- Google Cloud プロジェクト ID
- リージョンとゾーン
- ビルドパイプラインとデータ処理ワークフローで使用される Cloud Storage バケットの名前
環境変数はセッション間で保持されないため、チュートリアルを進めている間に Cloud Shell セッションがシャットダウンまたは切断された場合は、環境変数を再設定する必要があります。

Cloud Composer 環境を作成する

このチュートリアルでは、Cloud Composer 環境を作成します。

Cloud Shell で、Cloud Composer v2 API サービスエージェント拡張機能（roles/composer.ServiceAgentV2Ext）のロールを Cloud Composer サービスエージェントアカウントに追加します。
```
gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member serviceAccount:service-$PROJECT_NUMBER@cloudcomposer-accounts.iam.gserviceaccount.com \
    --role roles/composer.ServiceAgentV2Ext
```
Cloud Shell で Cloud Composer 環境を作成します。
```
gcloud composer environments create $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --image-version composer-2.0.14-airflow-2.2.5
```
注: Cloud Composer 環境のプロビジョニングには通常 15 分ほどかかりますが、1 時間程度かかることもあります。このプロセスが完了するまで待ってから、次の手順に進んでください。
スクリプトを実行して、Cloud Composer 環境の変数を設定します。これらの変数はデータ処理 DAG で必要になります。
```
cd ~/ci-cd-for-data-processing-workflow/env-setup
chmod +x set_composer_variables.sh
./set_composer_variables.sh
```
このスクリプトでは次の環境変数を設定します。
- Google Cloud プロジェクト ID
- リージョンとゾーン
- ビルドパイプラインとデータ処理ワークフローで使用される Cloud Storage バケットの名前

Cloud Composer 環境プロパティを抽出する

Cloud Composer は、Cloud Storage バケットを使用して DAG を保存します。DAG 定義ファイルをバケットに移動すると、Cloud Composer の読み取りがトリガーされ、自動的にファイルが読み取られます。Cloud Composer 環境を作成したときに、Cloud Composer 用の Cloud Storage バケットを作成しました。次の手順では、バケットの URL を抽出してから、DAG 定義を Cloud Storage バケットに自動的にデプロイするように CI / CD パイプラインを構成します。

Cloud Shell で、バケットの URL を環境変数としてエクスポートします。

export COMPOSER_DAG_BUCKET=$(gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --format="get(config.dagGcsPrefix)")

Cloud Storage バケットにアクセスできるようにするために、Cloud Composer が使用するサービスアカウントの名前をエクスポートします。

export COMPOSER_SERVICE_ACCOUNT=$(gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --format="get(config.nodeConfig.serviceAccount)")

Cloud Storage バケットを作成する

このセクションでは、次のデータを保存する一連の Cloud Storage バケットを作成します。

ビルドプロセスの中間ステップのアーティファクト。
データ処理ワークフローの入力ファイルと出力ファイル。
Dataflow ジョブのバイナリファイルを保存するためのステージングの場所。

Cloud Storage バケットを作成するには、以下の手順を実施します。

Cloud Shell で Cloud Storage バケットを作成し、Cloud Composer サービスアカウントにデータ処理ワークフローを実行する権限を付与します。
```
cd ~/ci-cd-for-data-processing-workflow/env-setup
chmod +x create_buckets.sh
./create_buckets.sh
```

Pub/Sub トピックを作成する

このセクションでは、本番環境ビルドパイプラインを自動的にトリガーするためにデータ処理ワークフロー統合テストから送信されたメッセージを受信する Pub/Sub トピックを作成します。

Google Cloud コンソールで、Pub/Sub の [トピック] ページに移動します。

トピックページに移動
[トピックを作成] をクリックします。
トピックを構成するには、次の手順を行います。
- [トピック ID] に「integration-test-complete-topic」と入力します。
- [デフォルトのサブスクリプションを追加] チェックボックスがオンになっていることを確認します。
- 残りのオプションはオフのままにします。
- [暗号化] で [Google が管理する暗号鍵] を選択します。
- [トピックを作成] をクリックします。

Cloud Source Repositories にソースコードを push する

このチュートリアルでは、バージョン管理に組み込む必要があるソースコードベースを 1 つ作成します。次の手順は、コードベースが開発され、時間の経過とともに変更されていく様子を表しています。変更がリポジトリに push されるたびに、ビルド、デプロイ、テストのパイプラインがトリガーされます。

Cloud Shell で、source-code フォルダを Cloud Source Repositories に push します。

gcloud source repos create $SOURCE_CODE_REPO
cp -r ~/ci-cd-for-data-processing-workflow/source-code ~/$SOURCE_CODE_REPO
cd ~/$SOURCE_CODE_REPO
git init
git remote add google \
    https://source.developers.google.com/p/$GCP_PROJECT_ID/r/$SOURCE_CODE_REPO
git add .
git commit -m 'initial commit'
git push google master

これらは、新しいディレクトリで Git を初期化し、コンテンツをリモートリポジトリに push する際の標準的なコマンドです。

Cloud Build パイプラインを作成する

このセクションでは、データ処理ワークフローをビルド、デプロイ、テストするビルドパイプラインを作成します。

Cloud Build サービスアカウントにアクセス権を付与する

Cloud Build は、Cloud Composer DAG をデプロイし、ワークフローをトリガーします。これは、Cloud Build サービスアカウントに追加のアクセス権を付与することによって可能になります。Cloud Composer で使用できるさまざまなロールの詳細については、アクセス制御のドキュメントをご覧ください。

Cloud Build ジョブが Cloud Composer で Airflow 変数を設定できるように、Cloud Shell で Cloud Build サービスアカウントに composer.admin ロールを追加します。
```
gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member=serviceAccount:$PROJECT_NUMBER@cloudbuild.gserviceaccount.com \
    --role=roles/composer.admin
```
Cloud Build ジョブが Cloud Composer でデータワークフローをトリガーできるように、Cloud Build サービスアカウントに composer.worker ロールを追加します。
```
gcloud projects add-iam-policy-binding $GCP_PROJECT_ID \
    --member=serviceAccount:$PROJECT_NUMBER@cloudbuild.gserviceaccount.com \
    --role=roles/composer.worker
```

ビルドとテストのパイプラインを作成する

ビルドとテストのパイプラインステップは YAML 構成ファイルで構成します。このチュートリアルでは、git、maven、gsutil、gcloud のビルド済みビルダーイメージを使用して、各ビルドステップのタスクを実行します。ビルド時に環境設定を定義するには、構成変数の置換を使用します。ソースコードリポジトリの場所は、変数の置換と Cloud Storage バケットの場所によって定義されます。JAR ファイル、テストファイル、DAG 定義をデプロイするときに、この情報が必要になります。

Cloud Shell で、ビルドパイプラインの構成ファイルを送信して Cloud Build 内にパイプラインを作成します。
```
cd ~/ci-cd-for-data-processing-workflow/source-code/build-pipeline
gcloud builds submit --config=build_deploy_test.yaml --substitutions=\
REPO_NAME=$SOURCE_CODE_REPO,\
_DATAFLOW_JAR_BUCKET=$DATAFLOW_JAR_BUCKET_TEST,\
_COMPOSER_INPUT_BUCKET=$INPUT_BUCKET_TEST,\
_COMPOSER_REF_BUCKET=$REF_BUCKET_TEST,\
_COMPOSER_DAG_BUCKET=$COMPOSER_DAG_BUCKET,\
_COMPOSER_ENV_NAME=$COMPOSER_ENV_NAME,\
_COMPOSER_REGION=$COMPOSER_REGION,\
_COMPOSER_DAG_NAME_TEST=$COMPOSER_DAG_NAME_TEST
```
このコマンドは、次の手順でビルドを実行するように Cloud Build に指示するものです。
1. WordCount の自己実行型 JAR ファイルをビルドしてデプロイします。
  1. ソースコードをチェックアウトします。
  2. WordCount Beam ソースコードを自己実行型 JAR ファイルにコンパイルします。
  3. この JAR ファイルを Cloud Storage に保存します。Cloud Composer はここからファイルを取得して WordCount 処理ジョブを実行できます。
2. データ処理ワークフローを Cloud Composer にデプロイして設定します。
  1. ワークフロー DAG で使用されるカスタムオペレータコードの単体テストを実行します。
  2. テストの入力ファイルとテストの参照ファイルを Cloud Storage にデプロイします。テストの入力ファイルは、WordCount 処理ジョブへの入力になります。テストの参照ファイルは、WordCount 処理ジョブの出力を検証する際の参照として使用されます。
  3. 新しくビルドされた JAR ファイルを指すように Cloud Composer 変数を設定します。
  4. ワークフロー DAG 定義を Cloud Composer 環境にデプロイします。
3. テスト環境でデータ処理ワークフローを実行し、テスト処理ワークフローをトリガーします。

ビルドとテストのパイプラインを検証する

ビルドファイルを送信したら、ビルドステップを検証します。

Google Cloud コンソールで [ビルド履歴] ページに移動し、過去および現在実行中のすべてのビルドのリストを表示します。

[ビルド履歴] ページに移動
現在実行中のビルドをクリックします。
[ビルドの詳細] ページで、そのビルドステップが上記のステップと一致していることを確認します。

[ビルドの詳細] ページで、ビルドが完了すると、ビルドの [ステータス] フィールドに「Build successful」と表示されます。
Cloud Shell で、WordCount のサンプル JAR ファイルが正しいバケットにコピーされたことを確認します。
```
gsutil ls gs://$DATAFLOW_JAR_BUCKET_TEST/dataflow_deployment*.jar
```
出力は次のようになります。
```
gs://…-composer-dataflow-source-test/dataflow_deployment_e88be61e-50a6-4aa0-beac-38d75871757e.jar
```
Cloud Composer のウェブインターフェースの URL を取得します。次の手順で使用するために、この URL をメモしておきます。
```
gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --format="get(config.airflowUri)"
```
前の手順の URL を使用して Cloud Composer UI に移動し、DAG が正しく実行されたことを確認します。[Runs] 列に情報が表示されない場合は、数分待ってからページを再読み込みしてください。
1. データ処理ワークフロー DAG test_word_count がデプロイされて実行モードになっていることを確認するには、[Runs] の下の薄い緑色の円にポインタを置き、[Running] と表示されていることを確認します。
2. 実行中のデータ処理ワークフローをグラフとして表示するには、薄い緑色の円をクリックし、[DAG Runs] ページで [Dag Id: test_word_count] をクリックします。
3. 現在の DAG の実行ステータスを更新するには、[Graph View] ページを再読み込みします。ワークフローが完了するまでに、通常 3〜5 分かかります。DAG の実行が正常に終了したことを確認するには、ポインタを各タスクの上に置き、ツールチップに「State: success」と表示されることを確認します。do_comparison という名前の最後から 2 番目のタスクは、参照ファイルに対してプロセスの出力を検証する統合テストです。
統合テストが完了すると、publish_test_complete という名前の最後のタスクで integration-test-complete-topic Pub/Sub トピックにメッセージが公開され、本番環境ビルドパイプラインがトリガーされます。
1. 公開されたメッセージに最新の JAR ファイルへの正しい参照が含まれていることを確認するには、デフォルトの integration-test-complete-topic-sub Pub/Sub サブスクリプションからメッセージを pull します。
2. Google Cloud コンソールで、[サブスクリプション] ページに移動します。
  
  [サブスクリプション] ページに移動
3. [integration-test-complete-topic-sub] をクリックし、[メッセージ] タブを選択して [Pull] をクリックします。
4. 出力例を以下に示します。

本番環境パイプラインを作成する

テスト処理ワークフローが正常に実行されたら、現在のバージョンのワークフローを本番環境に昇格させることができます。ワークフローを本番環境にデプロイするには、いくつかの方法があります。

手動。
テスト環境またはステージング環境ですべてのテストが正常に完了した場合に自動的にトリガーする。
スケジュール設定されたジョブによって自動的にトリガーする。

このチュートリアルでは、すべてのテストでテスト環境が正常に完了すると本番環境ビルドが自動的にトリガーされます。自動アプローチの詳細については、リリースエンジニアリングをご覧ください。

自動アプローチを実装する前に、本番環境に手動でデプロイして、本番環境用デプロイメントビルドを確認します。本番環境用デプロイメントビルドは次の手順で実行します。

テスト用バケットから本番環境用バケットに WordCount の JAR ファイルをコピーします。
本番環境用ワークフローの Cloud Composer 変数を設定して、新しく昇格する JAR ファイルを指すようにします。
本番環境用ワークフローの DAG 定義を Cloud Composer 環境にデプロイしてワークフローを実行します。

変数置換によって、本番環境にデプロイされる最新の JAR ファイルの名前が定義されます。本番環境の処理ワークフローで使用される Cloud Storage バケットに置換されます。本番環境用の Airflow ワークフローをデプロイする Cloud Build パイプラインを作成するには、以下の手順を実行します。

Cloud Shell で、JAR ファイル名の Cloud Composer 変数を出力して、最新の JAR ファイルのファイル名を読み取ります。

export DATAFLOW_JAR_FILE_LATEST=$(gcloud composer environments run $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION variables get -- \
    dataflow_jar_file_test 2>&1 | grep -i '.jar')

ビルドパイプライン構成ファイル deploy_prod.yaml, を使用して、Cloud Build でパイプラインを作成します。

cd ~/ci-cd-for-data-processing-workflow/source-code/build-pipeline
gcloud builds submit --config=deploy_prod.yaml --substitutions=\
REPO_NAME=$SOURCE_CODE_REPO,\
_DATAFLOW_JAR_BUCKET_TEST=$DATAFLOW_JAR_BUCKET_TEST,\
_DATAFLOW_JAR_FILE_LATEST=$DATAFLOW_JAR_FILE_LATEST,\
_DATAFLOW_JAR_BUCKET_PROD=$DATAFLOW_JAR_BUCKET_PROD,\
_COMPOSER_INPUT_BUCKET=$INPUT_BUCKET_PROD,\
_COMPOSER_ENV_NAME=$COMPOSER_ENV_NAME,\
_COMPOSER_REGION=$COMPOSER_REGION,\
_COMPOSER_DAG_BUCKET=$COMPOSER_DAG_BUCKET,\
_COMPOSER_DAG_NAME_PROD=$COMPOSER_DAG_NAME_PROD

本番環境パイプラインによって作成されたデータ処理ワークフローを検証する

Cloud Composer UI の URL を取得します。

gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION \
    --format="get(config.airflowUri)"

本番環境用データ処理ワークフロー DAG がデプロイされていることを確認するには、前の手順で取得した URL に移動し、prod_word_count DAG が DAG リストに含まれていることを確認します。
1. [DAGs] ページの [prod_word_count] 行で、[Trigger DAG] をクリックします。
Airflow ロゴをクリックするか、ページを再読み込みして DAG の実行ステータスを更新します。[Runs] 列の薄い緑色の円は、DAG が現在実行中であることを示します。円の上にポインタを置くと、[実行中] というツールチップが表示されます。
実行が完了したら、[DAG runs] 列の下の濃い緑色の円にポインタを置き、「Success」と表示されることを確認します。

Cloud Shell で、Cloud Storage バケットの結果ファイルを一覧表示します。

gsutil ls gs://$RESULT_BUCKET_PROD

出力は次のようになります。

gs://…-composer-result-prod/output-00000-of-00003
gs://…-composer-result-prod/output-00001-of-00003
gs://…-composer-result-prod/output-00002-of-00003

Cloud Build トリガーの作成

このセクションでは、ソースコードの変更をテストビルドプロセスにリンクし、テストパイプラインと本番環境ビルドパイプラインの間にリンクする Cloud Build トリガーを作成します。

テストビルドパイプライントリガーの構成

ソースリポジトリのマスターブランチに変更が push されたときに新しいビルドをトリガーする Cloud Build トリガーを設定します。

Google Cloud コンソールで、[ビルドトリガー] ページに移動します。

[ビルドトリガー] ページに移動
[トリガーを作成] をクリックします。
トリガー設定を構成するには、以下の手順を実施します。
- [名前] フィールドに「trigger-build-in-test-environment」と入力します。
- [リージョン] プルダウンで、[グローバル（非リージョン）] を選択します。
- [イベント] で [ブランチに push する] をクリックします。
- [ソース] で [data-pipeline-source] を選択します。
- [ブランチ名] フィールドに「master」と入力します。
- [構成] で [Cloud Build 構成ファイル（yaml または json）] をクリックします。
- [ロケーション] で [リポジトリ] をクリックします。
- [Cloud Build 構成ファイルの場所] フィールドに「build-pipeline/build_deploy_test.yaml」と入力します。

Cloud Shell で次のコマンドを実行して、ビルドに必要なすべての置換変数を取得します。後の手順で必要になるため、この値をすべてメモしておきます。

echo "_COMPOSER_DAG_BUCKET : ${COMPOSER_DAG_BUCKET}
_COMPOSER_DAG_NAME_TEST : ${COMPOSER_DAG_NAME_TEST}
_COMPOSER_ENV_NAME : ${COMPOSER_ENV_NAME}
_COMPOSER_INPUT_BUCKET : ${INPUT_BUCKET_TEST}
_COMPOSER_REF_BUCKET : ${REF_BUCKET_TEST}
_COMPOSER_REGION : ${COMPOSER_REGION}
_DATAFLOW_JAR_BUCKET : ${DATAFLOW_JAR_BUCKET_TEST}"

注: 出力の名前と値のペアが次の手順で使用されます。

[トリガー設定] ページの [高度な置換変数] で、変数を前の手順で取得した環境の値に置き換えます。次の値を一度に 1 つずつ追加して、名前と値のペアごとに [+ 項目を追加] をクリックします。
- _COMPOSER_DAG_BUCKET
- _COMPOSER_DAG_NAME_TEST
- _COMPOSER_ENV_NAME
- _COMPOSER_INPUT_BUCKET
- _COMPOSER_REF_BUCKET
- _COMPOSER_REGION
- _DATAFLOW_JAR_BUCKET
[作成] をクリックします。

本番環境のビルドパイプライントリガーの構成

テスト環境でテストに合格し、tests-complete Pub/Sub トピックにメッセージが公開されたときに本番環境のビルドをトリガーする Cloud Build トリガーを設定します。このトリガーには、本番環境パイプラインを実行する前にビルドを手動で承認する必要がある承認の手順が含まれています。

Google Cloud コンソールで、[ビルドトリガー] ページに移動します。

[ビルドトリガー] ページに移動
[トリガーを作成] をクリックします。
トリガー設定を構成するには、以下の手順を実施します。
- [名前] フィールドに「trigger-build-in-prod-environment」と入力します。
- [リージョン] プルダウンで、[グローバル（非リージョン）] を選択します。
- [イベント] で [Pub/Sub メッセージ] をクリックします。
- [サブスクリプション] で [integration-test-complete-topic] を選択します。
- [ソース] で [data-pipeline-source] を選択します。
- [リビジョン] で [ブランチ] を選択します。
- [ブランチ名] フィールドに「master」と入力します。
- [構成] で [Cloud Build 構成ファイル（yaml または json）] をクリックします。
- [ロケーション] で [リポジトリ] をクリックします。
- [Cloud Build 構成ファイルの場所] フィールドに「build-pipeline/deploy_prod.yaml」と入力します。

echo "_COMPOSER_DAG_BUCKET : ${COMPOSER_DAG_BUCKET}
_COMPOSER_DAG_NAME_PROD : ${COMPOSER_DAG_NAME_PROD}
_COMPOSER_ENV_NAME : ${COMPOSER_ENV_NAME}
_COMPOSER_INPUT_BUCKET : ${INPUT_BUCKET_PROD}
_COMPOSER_REGION : ${COMPOSER_REGION}
_DATAFLOW_JAR_BUCKET_PROD : ${DATAFLOW_JAR_BUCKET_PROD}
_DATAFLOW_JAR_BUCKET_TEST : ${DATAFLOW_JAR_BUCKET_TEST}"

注: 出力の名前と値のペアが次の手順で使用されます。

[トリガー設定] ページの [高度な置換変数] で、変数を前の手順で取得した環境の値に置き換えます。次の値を一度に 1 つずつ追加して、名前と値のペアごとに [+ 項目を追加] をクリックします。
- _COMPOSER_DAG_BUCKET
- _COMPOSER_DAG_NAME_PROD
- _COMPOSER_ENV_NAME
- _COMPOSER_INPUT_BUCKET
- _COMPOSER_REGION
- _DATAFLOW_JAR_BUCKET_PROD
- _DATAFLOW_JAR_BUCKET_TEST
- _DATAFLOW_JAR_FILE_LATEST = $(body.message.data)
[Approval] で [Require approval before build executes] をオンにします。
[作成] をクリックします。

トリガーをテストする

トリガーをテストするには、テスト入力ファイルに新しい単語を追加し、それに合わせてテスト参照ファイルも調整します。Cloud Source Repositories への commit push によってビルドパイプラインがトリガーされ、更新されたテストファイルを使用してデータ処理ワークフローが正しく実行されることを確認します。

Cloud Shell で、テストファイルの末尾にテスト用の単語を追加します。
```
echo "testword" >>  ~/$SOURCE_CODE_REPO/workflow-dag/support-files/input.txt
```
テスト入力ファイルで行った変更に合わせて、テスト結果の参照ファイル ref.txt を更新します。
```
echo "testword: 1" >>  ~/$SOURCE_CODE_REPO/workflow-dag/support-files/ref.txt
```

変更を commit して Cloud Source Repositories に push します。

cd ~/$SOURCE_CODE_REPO
git add .
git commit -m 'change in test files'
git push google master

Google Cloud コンソールの [履歴] ページに移動します。

[履歴] ページに移動
マスターブランチへの以前の push によって新しいテストビルドがトリガーされたことを確認するには、現在実行中のビルドで [Ref] 列に [master] と表示されていることを確認します。

Cloud Shell で、Cloud Composer のウェブインターフェースの URL を取得します。

gcloud composer environments describe $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION --format="get(config.airflowUri)"

ビルドが完了したら、前のコマンドの URL に移動して、test_word_count DAG が実行されていることを確認します。

DAG の実行が完了するまで待ちます。完了すると、[DAG runs] 列の薄い緑色の円が消えます。プロセスが完了するまでに通常 3〜5 分かかります。

Cloud Shell でテスト結果ファイルをダウンロードします。

mkdir ~/result-download
cd ~/result-download
gsutil cp gs://$RESULT_BUCKET_TEST/output* .

新しく追加した単語が結果ファイルのいずれかに含まれていることを確認します。
```
grep testword output*
```
出力は次のようになります。
```
output-00000-of-00003:testword: 1
```
Google Cloud コンソールの [履歴] ページに移動します。

[履歴] ページに移動
統合テストの完了によって新しい本番環境のビルドがトリガーされ、ビルドが承認待ちの状態であることを確認します。
ビルドの横にあるチェックボックスをオンにして、[承認] をクリックし、確認ボックスの [承認] をクリックして、本番環境のビルドパイプラインを実行します。
ビルドが完了したら、前のコマンドの URL に移動し、prod_word_count DAG を手動でトリガーして本番環境パイプラインを実行します。

クリーンアップ

このチュートリアルで使用したリソースについて、Google Cloud アカウントに課金されないようにするには、リソースを含むプロジェクトを削除するか、プロジェクトを維持して個々のリソースを削除します。

プロジェクトの削除

Google Cloud コンソールで、[リソースの管理] ページに移動します。
[リソースの管理] に移動
プロジェクトリストで、削除するプロジェクトを選択し、[削除] をクリックします。
ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

個々のリソースの削除

このチュートリアルで使用したプロジェクトをそのまま使用する場合は、次の手順を実行して、このチュートリアルで作成したリソースを削除してください。

Cloud Build トリガーを削除するには、次の手順を実施します。
1. Google Cloud コンソールの [トリガー] ページに移動します。
  
  [トリガー] ページに移動
2. 作成したトリガーの横にある [さらに表示] をクリックし、[削除] をクリックします。

Cloud Shell で Cloud Composer 環境を削除します。

gcloud -q composer environments delete $COMPOSER_ENV_NAME \
    --location $COMPOSER_REGION

Cloud Storage バケットとそのファイルを削除します。

gsutil -m rm -r gs://$DATAFLOW_JAR_BUCKET_TEST \
    gs://$INPUT_BUCKET_TEST \
    gs://$REF_BUCKET_TEST \
    gs://$RESULT_BUCKET_TEST \
    gs://$DATAFLOW_STAGING_BUCKET_TEST \
    gs://$DATAFLOW_JAR_BUCKET_PROD \
    gs://$INPUT_BUCKET_PROD \
    gs://$RESULT_BUCKET_PROD \
    gs://$DATAFLOW_STAGING_BUCKET_PROD

Pub/Sub トピックとデフォルトのサブスクリプションを削除するには、Cloud Shell で次のコマンドを実行します。
```
gcloud pubsub topics delete integration-test-complete-topic
gcloud pubsub subscriptions delete integration-test-complete-topic-sub
```

リポジトリを削除します。

gcloud -q source repos delete $SOURCE_CODE_REPO

作成したファイルとフォルダを削除します。

rm -rf ~/ci-cd-for-data-processing-workflow
rm -rf ~/$SOURCE_CODE_REPO
rm -rf ~/result-download

次のステップ

Cloud Build を使用した GitOps スタイルの継続的デリバリーの詳細を見る。
Cloud Composer を使用したインフラストラクチャの自動化の詳細を見る。
一般的な Dataflow のユースケースパターンの詳細を確認する。
リリースエンジニアリングの詳細を見る。
Google Cloud に関するリファレンスアーキテクチャ、図、ベストプラクティスを確認する。Cloud Architecture Center を確認します。

データ処理ワークフロー用の CI / CD パイプラインの設定

デプロイ アーキテクチャ

CI / CD パイプライン

データ処理ワークフロー

テスト

目標

費用

始める前に

サンプルコード

環境設定

Cloud Composer 環境を作成する

Cloud Composer 環境プロパティを抽出する

Cloud Storage バケットを作成する

Pub/Sub トピックを作成する

Cloud Source Repositories にソースコードを push する

Cloud Build パイプラインを作成する

Cloud Build サービス アカウントにアクセス権を付与する

ビルドとテストのパイプラインを作成する

ビルドとテストのパイプラインを検証する

本番環境パイプラインを作成する

本番環境パイプラインによって作成されたデータ処理ワークフローを検証する

Cloud Build トリガーの作成

テストビルド パイプライン トリガーの構成

本番環境のビルド パイプライン トリガーの構成

トリガーをテストする

クリーンアップ

プロジェクトの削除

個々のリソースの削除

次のステップ

デプロイアーキテクチャ

Cloud Build サービスアカウントにアクセス権を付与する

テストビルドパイプライントリガーの構成

本番環境のビルドパイプライントリガーの構成