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

このチュートリアルでは、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 パイプラインは次のステップで構成されます。

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

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

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

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

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

  1. デベロッパーがコードの変更を Cloud Source Repositories に commit します。
  2. コードの変更によって Cloud Build でテストビルドがトリガーされます。
  3. Cloud Build が自己実行型 JAR ファイルをビルドし、それを Cloud Storage のテスト用 JAR バケットにデプロイします。
  4. Cloud Build がテストファイルを Cloud Storage のテストファイル バケットにデプロイします。
  5. Cloud Build が、新しくデプロイされた JAR ファイルを参照するように Cloud Composer の変数を設定します。
  6. Cloud Build が、データ処理ワークフローの有向非巡回グラフ(DAG)をテストし、Cloud Storage の Cloud Composer バケットにデプロイします。
  7. このワークフローの DAG ファイルが Cloud Composer にデプロイされます。
  8. Cloud Build が、新しくデプロイされたデータ処理ワークフローの実行をトリガーします。

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

  1. デベロッパーが Cloud Build の本番環境デプロイメント パイプラインを手動で実行します。
  2. Cloud Build が、最新の自己実行型 JAR ファイルを Cloud Storage のテスト用 JAR バケットから本番環境用 JAR バケットにコピーします。
  3. Cloud Build が、本番環境用データ処理ワークフロー DAG をテストし、Cloud Storage の Cloud Composer バケットにデプロイします。
  4. 本番環境用のワークフロー 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 ステップのデータ処理ワークフロー。

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

  1. Dataflow で WordCount データ処理を実行します。
  2. WordCount プロセスの出力ファイルをダウンロードします。WordCount プロセスは次の 3 つのファイルを出力します。

    • download_result_1
    • download_result_2
    • download_result_3
  3. download_ref_string という名前の参照ファイルをダウンロードします。

  4. この参照ファイルと照らし合わせて結果を検証します。この統合テストでは、3 つの結果すべてを集計して、集計結果を参照ファイルと比較します。

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

テスト

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

目標

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

費用

このチュートリアルでは、課金対象である次の Google Cloud コンポーネントを使用します。

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

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

始める前に

  1. Google Cloud アカウントにログインします。Google Cloud を初めて使用する場合は、アカウントを作成して、実際のシナリオでの Google プロダクトのパフォーマンスを評価してください。新規のお客様には、ワークロードの実行、テスト、デプロイができる無料クレジット $300 分を差し上げます。
  2. Google Cloud Console の [プロジェクト セレクタ] ページで、Google Cloud プロジェクトを選択または作成します。

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

  3. Cloud プロジェクトに対して課金が有効になっていることを確認します。詳しくは、プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。

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

    API を有効にする

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

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

  6. Cloud プロジェクトに対して課金が有効になっていることを確認します。詳しくは、プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。

  7. 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 Console の下部にウィンドウとして表示されます。

  1. Cloud Console で Cloud Shell を開きます。

    Cloud Shell を開く

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

    git clone https://github.com/GoogleCloudPlatform/ci-cd-for-data-processing-workflow.git
    
  3. スクリプトを実行して環境変数を設定します。

    cd ~/ci-cd-for-data-processing-workflow/env-setup
    source set_env.sh
    

    このスクリプトでは次の環境変数を設定します。

    • Google Cloud プロジェクト ID
    • リージョンとゾーン
    • ビルド パイプラインとデータ処理ワークフローで使用される Cloud Storage バケットの名前

    環境変数はセッション間で保持されないため、チュートリアルを進めている間に Cloud Shell セッションがシャットダウンまたは切断された場合は、環境変数を再設定する必要があります。

Cloud Composer 環境を作成する

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

  1. 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
    
  2. Cloud Shell で Cloud Composer 環境を作成します。

    gcloud composer environments create $COMPOSER_ENV_NAME \
        --location $COMPOSER_REGION \
        --image-version composer-2.0.14-airflow-2.2.5
    
  3. スクリプトを実行して、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 パイプラインを構成します。

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

    export COMPOSER_DAG_BUCKET=$(gcloud composer environments describe $COMPOSER_ENV_NAME \
        --location $COMPOSER_REGION \
        --format="get(config.dagGcsPrefix)")
    
  2. 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
    

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 で使用できるさまざまなロールの詳細については、アクセス制御のドキュメントをご覧ください。

  1. 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
    
  2. 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 構成ファイルで構成します。このチュートリアルでは、gitmavengsutilgcloud のビルド済みビルダー イメージを使用して、各ビルドステップのタスクを実行します。ビルド時に環境設定を定義するには、構成変数の置換を使用します。ソースコード リポジトリの場所は、変数の置換と 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. テスト環境でデータ処理ワークフローを実行し、テスト処理ワークフローをトリガーします。

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

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

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

    [ビルド履歴] ページに移動

  2. 現在実行中のビルドをクリックします。

  3. [ビルドの詳細] ページで、そのビルドステップが上記のステップと一致していることを確認します。

    ビルドステップの詳細。

    [ビルドの詳細] ページで、ビルドが完了すると、ビルドの [ステータス] フィールドに「Build successful」と表示されます。

  4. 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
    
  5. Cloud Composer のウェブ インターフェースの URL を取得します。次の手順で使用するために、この URL をメモしておきます。

    gcloud composer environments describe $COMPOSER_ENV_NAME \
        --location $COMPOSER_REGION \
        --format="get(config.airflowUri)"
    
  6. 前の手順の URL を使用して Cloud Composer UI に移動し、DAG が正しく実行されたことを確認します。[DAG Runs] 列に情報が表示されない場合は、数分待ってからページを再読み込みしてください。

    1. データ処理ワークフロー DAG test_word_count がデプロイされて実行モードになっていることを確認するには、[DAG Runs] の下の薄い緑色の円にポインタを置き、「Running」と表示されることを確認します。

      実行中の DAG 処理ステータス。

    2. 実行中のデータ処理ワークフローをグラフとして表示するには、薄い緑色の円をクリックし、[DAG Runs] ページで [Dag Id: test_word_count] をクリックします。

    3. 現在の DAG の実行ステータスを更新するには、[Graph View] ページを再読み込みします。ワークフローが完了するまでに、通常 3〜5 分かかります。DAG の実行が正常に終了したことを確認するには、ポインタを各タスクの上に置き、ツールチップに「State: success」と表示されることを確認します。最後のタスク do_comparison は、プロセスの出力を参照ファイルと照らし合わせて検証する統合テストです。

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

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

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

自動アプローチはこのチュートリアルの対象範囲外です。詳しくは、リリース エンジニアリングをご覧ください。

このチュートリアルでは、Cloud Build の本番環境用デプロイメント ビルドを実行して本番環境に手動でデプロイします。本番環境用デプロイメント ビルドは次の手順で実行します。

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

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

  1. 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')
    
  2. ビルド パイプライン構成ファイル 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
    

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

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

    gcloud composer environments describe $COMPOSER_ENV_NAME \
        --location $COMPOSER_REGION \
        --format="get(config.airflowUri)"
    
  2. 本番環境用データ処理ワークフロー DAG がデプロイされていることを確認するには、前の手順で取得した URL に移動し、prod_word_count DAG が DAG リストに含まれていることを確認します。

    1. [DAGs] ページの [prod_word_count] 行で、[Trigger Dag] をクリックします。

      DAG をトリガーする再生アイコン。

    2. [Confirmation] ダイアログで [Confirm] をクリックします。

  3. DAG の実行ステータスを更新するには、ページを再読み込みします。本番環境用データ処理ワークフロー DAG がデプロイされ実行モードになっていることを確認するには、[DAG Runs] の下の薄い緑色の円にポインタを置き、「Running」と表示されることを確認します。

    実行中の DAG Runs ステータス。

  4. 実行が完了したら、[DAG runs] 列の下の濃い緑色の円にポインタを置き、「Success」と表示されることを確認します。

    成功時の DAG の実行ステータス。

  5. 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
    

ビルドトリガーを構成する

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

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

    echo "_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}"
    
  2. Cloud Console で、[ビルドトリガー] ページに移動します。

    [ビルドトリガー] ページに移動

  3. [トリガーを作成] をクリックします。

  4. [Cloud Source Repositories] をクリックし、[続行] をクリックします。

  5. [data-pipeline-source] をクリックして、[続行] をクリックします。

  6. トリガー設定を構成するには、以下の手順を実施します。

    • [名前] フィールドに「trigger-build-in-test-environment」と入力します。
    • [トリガーのタイプ] で [ブランチ] をクリックします。
    • [ブランチ(正規表現)] フィールドに master と入力します。
    • [構成] で [Cloud Build 構成ファイル(yaml または json)] をクリックします。
    • [Cloud Build 構成ファイルの場所] フィールドに「build-pipeline/build_deploy_test.yaml」と入力します。
  7. [トリガー設定] ページで、変数を前の手順で取得した環境の値に置き換えます。次の値を一度に 1 つずつ追加して、名前と値のペアごとに [+ 項目を追加] をクリックします。

    • _DATAFLOW_JAR_BUCKET
    • _COMPOSER_INPUT_BUCKET
    • _COMPOSER_REF_BUCKET
    • _COMPOSER_DAG_BUCKET
    • _COMPOSER_ENV_NAME
    • _COMPOSER_REGION
    • _COMPOSER_DAG_NAME_TEST

      名前と値のペアのマッピング。

  8. [トリガーを作成] をクリックします。

トリガーをテストする

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

  1. Cloud Shell で、テストファイルの末尾にテスト用の単語を追加します。

    echo "testword" >>  ~/$SOURCE_CODE_REPO/workflow-dag/support-files/input.txt
    
  2. テスト入力ファイルで行った変更に合わせて、テスト結果の参照ファイル ref.txt を更新します。

    echo "testword: 1" >>  ~/$SOURCE_CODE_REPO/workflow-dag/support-files/ref.txt
    
  3. 変更を commit して Cloud Source Repositories に push します。

    cd ~/$SOURCE_CODE_REPO
    git add .
    git commit -m 'change in test files'
    git push google master
    
  4. Cloud Console で [履歴] ページに移動します。

    [履歴] ページに移動

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

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

    gcloud composer environments describe $COMPOSER_ENV_NAME \
        --location $COMPOSER_REGION --format="get(config.airflowUri)"
    
  7. ビルドが完了したら、前のコマンドの URL に移動して、test_word_count DAG が実行されていることを確認します。

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

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

    mkdir ~/result-download
    cd ~/result-download
    gsutil cp gs://$RESULT_BUCKET_TEST/output* .
    
  9. 新しく追加した単語が結果ファイルのいずれかに含まれていることを確認します。

    grep testword output*
    

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

    output-00000-of-00003:testword: 1
    

クリーンアップ

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

プロジェクトの削除

  1. Google Cloud Console で、[リソースの管理] ページに移動します。

    [リソースの管理] に移動

  2. プロジェクト リストで、削除するプロジェクトを選択し、[削除] をクリックします。
  3. ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

個々のリソースの削除

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

  1. Cloud Build トリガーを削除するには、次の手順を実施します。

    1. Cloud Console で、[トリガー] ページに移動します。

      [トリガー] ページに移動

    2. 作成したトリガーの横にある [さらに表示] をクリックし、[削除] をクリックします。

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

    gcloud -q composer environments delete $COMPOSER_ENV_NAME \
        --location $COMPOSER_REGION
    
  3. 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
    
  4. リポジトリを削除します。

    gcloud -q source repos delete $SOURCE_CODE_REPO
    
  5. 作成したファイルとフォルダを削除します。

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

次のステップ