概要

このラボでは、次の作業を行います。

• ブランチがあるパイプラインを実装する
• 書き込む前にデータをフィルタする
• カスタム コマンドライン パラメータをパイプラインに追加する
• カスタム パイプラインをカスタム Dataflow Flex テンプレートに変換する
• Dataflow Flex テンプレートを実行する

前提条件:

• Python に関する基本的な知識

前のラボでは、基本的な抽出-変換-読み込みの順次実行パイプラインを作成し、対応する Dataflow テンプレートを使用して Google Cloud Storage 上にバッチ データストレージを取り込みました。このパイプラインは、以下に示す変換のシーケンスで構成されています。

しかし多くの場合、パイプラインはこのように単純な構造ではありません。このラボでは、より高度な連続的でないパイプラインを構築します。

今回のユースケースではリソース消費量を最適化します。プロダクトによってリソースの利用状況は異なります。また、一つの企業内でもすべてのデータが同じように使われるわけではなく、たとえば分析ワークロードで定期的にクエリされるデータもあれば、復元にのみ使用されるデータもあります。このラボでは、最初のラボで作成したパイプラインのリソース消費量を最適化するために、アナリストが使用するデータのみを BigQuery に保存し、他のデータは低コストで耐久性の高いストレージ サービスである Google Cloud Storage の Coldline Storage にアーカイブします。

設定と要件

Qwiklabs の設定

各ラボでは、新しい Google Cloud プロジェクトとリソースセットを一定時間無料で利用できます。

1. Qwiklabs にシークレット ウィンドウでログインします。

2. ラボのアクセス時間（例: 1:15:00）に注意し、時間内に完了できるようにしてください。
   一時停止機能はありません。必要な場合はやり直せますが、最初からになります。

3. 準備ができたら、[ラボを開始] をクリックします。

4. ラボの認証情報（ユーザー名とパスワード）をメモしておきます。この情報は、Google Cloud Console にログインする際に使用します。

5. [Google Console を開く] をクリックします。

6. [別のアカウントを使用] をクリックし、このラボの認証情報をコピーしてプロンプトに貼り付けます。
   他の認証情報を使用すると、エラーが発生したり、料金の請求が発生したりします。

7. 利用規約に同意し、再設定用のリソースページをスキップします。

プロジェクトの権限を確認する

Google Cloud で作業を開始する前に、Identity and Access Management（IAM）内で適切な権限がプロジェクトに付与されていることを確認する必要があります。

1. Google Cloud コンソールのナビゲーション メニュー（）で、[IAM と管理] > [IAM] を選択します。

2. Compute Engine のデフォルトのサービス アカウント {project-number}-compute@developer.gserviceaccount.com が存在し、編集者のロールが割り当てられていることを確認します。アカウントの接頭辞はプロジェクト番号で、ナビゲーション メニュー > [Cloud の概要] > [ダッシュボード] から確認できます。

注: アカウントが IAM に存在しない場合やアカウントに編集者のロールがない場合は、以下の手順に沿って必要なロールを割り当てます。

1. Google Cloud コンソールのナビゲーション メニューで、[Cloud の概要] > [ダッシュボード] をクリックします。
2. プロジェクト番号（例: 729328892908）をコピーします。
3. ナビゲーション メニューで、[IAM と管理] > [IAM] を選択します。
4. ロールの表の上部で、[プリンシパル別に表示] の下にある [アクセスを許可] をクリックします。
5. [新しいプリンシパル] に次のように入力します。

{project-number}-compute@developer.gserviceaccount.com

6. {project-number} はプロジェクト番号に置き換えてください。
7. [ロール] で、[Project]（または [基本]）> [編集者] を選択します。
8. [保存] をクリックします。

Jupyter ノートブック ベースの開発環境の設定

このラボでは、すべてのコマンドをノートブックのターミナルで実行します。

1. Google Cloud コンソールのナビゲーション メニューで、[Vertex AI] > [Workbench] をクリックします。

2. [Notebooks API を有効にする] をクリックします。

3. [Workbench] ページで [ユーザー管理のノートブック] を選択し、[新規作成] をクリックします。

4. 表示された [新しいインスタンス] ダイアログ ボックスで、リージョンを に、ゾーンを に設定します。

5. [環境] で [Apache Beam] を選択します。

6. ダイアログ ボックスの下部にある [作成] をクリックします。

注: 環境の完全なプロビジョニングには 3～5 分かかる場合があります。処理が完了するまでお待ちください。 注: [Notebook API を有効にする] をクリックして Notebooks API を有効にします。

7. 環境の準備が完了したら、ノートブック名の横にある [JupyterLab を開く] をクリックします。これにより、使用環境がブラウザの新しいタブで開きます。

8. 次に、[ターミナル] をクリックします。これにより、このラボのすべてのコマンドを実行できるターミナルが開きます。

コード リポジトリをダウンロードする

このラボで使用するコード リポジトリをダウンロードします。

1. 開いたターミナルで、次のコマンドを入力します。

git clone https://github.com/GoogleCloudPlatform/training-data-analyst cd /home/jupyter/training-data-analyst/quests/dataflow_python/

2. ノートブック環境の左側パネルのファイル ブラウザに、training-data-analyst リポジトリが追加されます。

3. クローン リポジトリ /training-data-analyst/quests/dataflow_python/ に移動します。ラボごとに、1 つのフォルダが表示されます。このフォルダはさらに、完成させるコードが格納される lab サブフォルダと、ヒントが必要な場合に完全に機能するサンプルを参照できる solution サブフォルダとに分けられています。

注: 編集のためにファイルを開くには、目的のファイルに移動してクリックします。ファイルが開き、コードを追加または変更できます。

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 ノートブック インスタンスを作成し、コース リポジトリのクローンを作成する

ラボ パート 1: 分岐するパイプラインの作成

このパートでは、Google Cloud Storage と BigQuery の両方にデータを書き込む、分岐するパイプラインを作成します。

複数の変換が同じ PCollection を処理する

分岐するパイプラインを作成する方法の一つは、2 つの異なる変換を同じ PCollection に適用することにより、2 つの異なる PCollection を作成することです。

[PCollection1] = [Initial Input PCollection] | [A Transform] [PCollection2] = [Initial Input PCollection] | [A Different Transform]

分岐するパイプラインを実装する

これ以降のセクションでヒントが必要な場合は、こちらのソリューションをご利用ください。

タスク 1: Cloud Storage に書き込むブランチを追加する

このタスクを完了するには、既存のパイプラインを変更し、Cloud Storage に書き込むブランチを追加します。

適切なラボを開く

IDE 環境に新しいターミナルをまだ作成していない場合は作成し、次のコマンドをコピーして貼り付けます。

# ディレクトリをラボに移動する cd 2_Branching_Pipelines/lab export BASE_DIR=$(pwd)

仮想環境と依存関係を設定する

実際のパイプライン コードの編集を開始できるようにするには、前もって必要な依存関係がインストールされていることを確認する必要があります。

IDE 環境で以前に開いていたターミナルに戻り、このラボの作業用に仮想環境を作成します。

sudo apt-get install -y python3-venv python3 -m venv df-env source df-env/bin/activate

次に、パイプライン中で実行する必要があるパッケージをインストールします。

python3 -m pip install -q --upgrade pip setuptools wheel python3 -m pip install apache-beam[gcp]

最後に、Dataflow API が有効になっていることを確認します。

gcloud services enable dataflow.googleapis.com

データ環境を設定する

# GCS バケットと BQ データセットを作成する cd $BASE_DIR/../.. source create_batch_sinks.sh # イベント データフローを生成する source generate_batch_events.sh # 練習バージョンのコードを含むディレクトリに移動する cd $BASE_DIR

IDE 中で 2_Branching_Pipelines/labs/ にある my_pipeline.py を開きます。パイプラインの本体部分を定義している run() メソッドまで下にスクロールします。現在は次のような内容です。

(p | 'ReadFromGCS' >> beam.io.ReadFromText(input) | 'ParseJson' >> beam.Map(parse_json) | 'WriteToBQ' >> beam.io.WriteToBigQuery( output, schema=table_schema, create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE ) )

このコードを変更し、各要素が json から dict に変換される前に、textio.WriteToText を使用して Cloud Storage への書き込みを行う新しい分岐変換を追加します。

これ以降のセクションでヒントが必要な場合は、こちらにあるソリューションをご覧ください。

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 データ環境を設定する

タスク 2: フィールドでデータをフィルタする

この時点では、すべてのデータが 2 回保存されるため、新しいパイプラインでもリソースの消費量は減りません。リソースの消費量を改善するには、重複するデータの量を減らす必要があります。Google Cloud Storage バケットの使用目的は、アーカイブおよびバックアップ ストレージとしての機能なので、すべてのデータを保存する必要があります。一方、BigQuery には必ずしもすべてのデータを送る必要はありません。

たとえば、データ アナリストが頻繁に確認する対象が、ウェブサイトでユーザーがどのリソースにアクセスしているか、そしてそのアクセス パターンが地域と時間によってどのように異なるかであると仮定します。これに必要なフィールドは一部のみです。すでに JSON をパースして辞書型に変換しているので、pop メソッドを使用して容易に Python の呼び出し可能オブジェクトからフィールドを除くことができます。

def drop_field(element): element.pop('field_name') return element

このタスクを完了するには、Python の呼び出し可能オブジェクトを beam.Map とともに使用して、アナリストが BigQuery で使用しないフィールド user_agent を除きます。

タスク 3: 要素でデータをフィルタする

Apache Beam にはフィルタリングの方法が数多くあります。Python 辞書形式の PCollection を取り扱っているため、最も簡単なのはブール値を返す関数である（匿名の）ラムダ関数をフィルタとして利用する方法で、beam.Filter とともに使用します。次に例を示します。

purchases | beam.Filter(lambda element : element['cost_cents'] > 20*100)

このタスクを完了するには、パイプラインに beam.Filter 変換を追加します。どのような条件でもフィルタできますが、たとえば num_bytes が 120 以上である行を除くということを試してみると良いでしょう。

タスク 4: カスタム コマンドライン パラメータを追加する

パイプラインには現在、入力のパスや BigQuery のテーブルの場所など、多くのパラメータがハードコードされています。Cloud Storage の任意の JSON ファイルを読み取ることができれば、パイプラインがさらに便利になります。この機能を追加するには、一連のコマンドライン パラメータへの追加が必要です。

現在は、コマンドライン引数の読み込みと解析に ArgumentParser を使用しています。そして、パイプライン作成時に指定した PipelineOptions() オブジェクトに引数を渡します。

parser = argparse.ArgumentParser(description='...') # 引数を定義して解析する options = PipelineOptions() # オプションからオプション値を設定する p = beam.Pipeline(options=options)

PipelineOptions を使用して、ArgumentParser で読み込んだオプションを解釈します。このパーサに新しいコマンドライン引数を追加するには、以下の構文を使用します。

parser.add_argument('--argument_name', required=True, help='Argument description')

コードでコマンドライン引数にアクセスするには、引数を解析した結果できる辞書のフィールドを参照します。

opts = parser.parse_args() arg_value = opts.arg_name

このタスクを完了するには、入力パス、Google Cloud Storage 出力パス、BigQuery テーブル名を示すコマンドライン パラメータを追加して、定数の代わりにこれらのパラメータにアクセスするようにパイプライン コードを更新します。

タスク 5: パイプラインに NULLABLE フィールドを追加する

お気づきかもしれませんが、前回のラボで作成した BigQuery テーブルには、すべてのフィールドを REQUIRED とする次のようなスキーマがありました。

データが存在しない NULLABLE フィールドがある Apache Beam スキーマを作成し、パイプライン実行自体および結果の BigQuery テーブルにもそれを反映したスキーマで適用するのが良いでしょう。

次のように、NULL 値を許容したいフィールドに新しいプロパティ mode を追加して JSON BigQuery スキーマを更新します。

{ "name": "field_name", "type": "STRING", "mode": "NULLABLE" }

このタスクを完成するには、BigQuery スキーマの lat フィールドと lon フィールドを null 値許容に設定します。

タスク 6: コマンドラインでパイプラインを実行する

このタスクを完了するには、コマンドラインでパイプラインを実行して適切なパラメータを渡します。生成される BigQuery スキーマの NULLABLE フィールドを忘れずにメモしておいてください。コードは次のようになります。

# 環境変数を設定する export PROJECT_ID=$(gcloud config get-value project) export REGION={{{project_0.startup_script.lab_region|Region}}} export BUCKET=gs://${PROJECT_ID} export COLDLINE_BUCKET=${BUCKET}-coldline export PIPELINE_FOLDER=${BUCKET} export RUNNER=DataflowRunner export INPUT_PATH=${PIPELINE_FOLDER}/events.json export OUTPUT_PATH=${PIPELINE_FOLDER}-coldline/pipeline_output export TABLE_NAME=${PROJECT_ID}:logs.logs_filtered cd $BASE_DIR python3 my_pipeline.py \ --project=${PROJECT_ID} \ --region=${REGION} \ --stagingLocation=${PIPELINE_FOLDER}/staging \ --tempLocation=${PIPELINE_FOLDER}/temp \ --runner=${RUNNER} \ --inputPath=${INPUT_PATH} \ --outputPath=${OUTPUT_PATH} \ --tableName=${TABLE_NAME} パイプラインが問題なく構築されているのにコードやデータフロー サービスの設定ミスによりエラーが大量に発生する場合には、RUNNER を「DirectRunner」に戻してローカルに実行すると、速くフィードバックを得ることができます。今回のケースはデータセットが小規模で、DirectRunner がサポートしている機能のみを使用しているため、この手法が有効です。

タスク 7: パイプラインの結果を確認する

Cloud Dataflow の [ジョブ] ページに移動して、実行中のジョブを確認します。グラフは次のようになっているはずです。

Filter 関数を表すノード（上の図では FilterFn）をクリックします。右側に表示されたパネルで、入力として追加された要素が出力として書き込まれた要素よりも多いことが確認できます。

次に Cloud Storage への書き込みを表すノードをクリックします。すべての要素が書き込まれているので、この数字は Filter 関数への入力の要素数と一致しているはずです。

パイプラインが終了したら、テーブルに対してクエリを実行して BigQuery の結果を確認します。テーブル内のレコード数は Filter 関数で出力された要素の数と一致しているはずです。

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 コマンドラインでパイプラインを実行する

ラボ パート 2: カスタム Dataflow テンプレート

コマンドライン パラメータを受け入れるパイプラインは、パラメータがハードコードされたパイプラインよりもはるかに便利です。しかしその実行には開発環境の作成が必要です。さまざまなユーザーによる再実行や、多種多様なコンテキストでの再実行が想定されるパイプラインには、さらに便利な選択肢があります。それは、Dataflow テンプレートを使うことです。

Google Cloud Platform には、あらかじめ作成された Dataflow テンプレートが多数用意されていて、こちらで確認できます。その中にこのラボのパイプラインと同じ動作をするテンプレートはありません。しかし、ラボのこのパートで、パイプラインを（従来のカスタム テンプレートではなく）新しいカスタム Dataflow Flex テンプレートに変換できます。

パイプラインをカスタム Flex Dataflow テンプレートに変換するには、コードだけでなく依存関係もパッケージ化する Docker コンテナ、ビルド対象のコードを記述する Dockerfile、実際のジョブの作成のためにランタイムに実行する基盤のコンテナをビルドする Cloud Build、ジョブ パラメータを記述するメタデータ ファイルを使用する必要があります。

タスク 1: カスタム Dataflow Flex テンプレートのコンテナ イメージを作成する

1. まず、Kaniko キャッシュの使用をデフォルトで有効にします。Kaniko は、コンテナビルドのアーティファクトをキャッシュに保存します。このオプションを使用すると、以降のビルド時間を短縮できます。また、pip3 freeze を使用して、現環境で使用しているパッケージとそのバージョンを記録します。

gcloud config set builds/use_kaniko True

2. 次に、Dockerfile を作成します。これにより、使用する必要があるコードおよび依存関係を指定します。

   a. このタスクを完了するには、IDE のファイル エクスプローラで dataflow_python/2_Branching_Pipelines/lab フォルダに新しいファイルを作成します。

   b. 新しいファイルを作成するには、[ファイル] >> [新規] >> [テキスト ファイル] の順にクリックします。

   c. ファイル名を Dockerfile に変更します。ファイル名を右クリックすると変更できます。

   d. 編集パネルで Dockerfile を開きます。ファイルをクリックすると開きます。

   e. 下のコードを Dockerfile ファイルにコピーして保存します。

FROM gcr.io/dataflow-templates-base/python3-template-launcher-base ARG WORKDIR=/dataflow/template RUN mkdir -p ${WORKDIR} WORKDIR ${WORKDIR} RUN apt-get update && apt-get install -y libffi-dev && rm -rf /var/lib/apt/lists/* COPY my_pipeline.py . ENV FLEX_TEMPLATE_PYTHON_PY_FILE="${WORKDIR}/my_pipeline.py" RUN python3 -m pip install apache-beam[gcp]==2.25.0

3. 最後に、Cloud Build を使用してコンテナ イメージをビルドします。

export TEMPLATE_IMAGE="gcr.io/$PROJECT_ID/dataflow/my_pipeline:latest" gcloud builds submit --tag $TEMPLATE_IMAGE .

4. コンテナをビルドして push するのに数分かかります。

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 カスタム Dataflow Flex テンプレートのコンテナ イメージを作成する

タスク 2: Flex テンプレートを作成してステージングする

1. テンプレートを実行するには、SDK 情報やメタデータなど、ジョブの実行に必要なすべての情報を含むテンプレート仕様ファイルを Cloud Storage に作成する必要があります。

   a. IDE のファイル エクスプローラで dataflow_python/2_Branching_Pipelines/lab フォルダに新しいファイルを作成します。

   b. 新しいファイルを作成するには、[ファイル] >> [新規] >> [テキスト ファイル] の順にクリックします。

   c. ファイル名を metadata.json に変更します。ファイル名を右クリックすると変更できます。

   d. 編集パネルで metadata.json ファイルを開きます。ファイルを開くには、metadata.json ファイルを右クリックし、[アプリで開く] >> [エディタ] の順に選択します。

   e. このタスクを完了するには、パイプラインで想定されるすべての入力パラメータを記述する次の形式で、metadata.json ファイルを作成します。必要な場合は、こちらでソリューションを参照してください。独自の正規表現チェックを記述する必要があります。おすすめの方法ではありませんが、".*" はあらゆる入力に一致します。

{ "name": "My Branching Pipeline", "description": "A branching pipeline that writes raw to GCS Coldline, and filtered data to BQ", "parameters": [ { "name": "inputPath", "label": "Input file path.", "helpText": "Path to events.json file.", "regexes": [ ".*\\.json" ] }, { "name": "outputPath", "label": "Output file location", "helpText": "GCS Coldline Bucket location for raw data", "regexes": [ "gs:\\/\\/[a-zA-z0-9\\-\\_\\/]+" ] }, { "name": "tableName", "label": "BigQuery output table", "helpText": "BigQuery table spec to write to, in the form 'project:dataset.table'.", "regexes": [ "[^:]+:[^.]+[.].+" ] } ] }

2. 次に、実際のテンプレートをビルドしてステージングします。

export TEMPLATE_PATH="gs://${PROJECT_ID}/templates/mytemplate.json" # テンプレートをビルドして GCS にアップロードする # gcloud のベータ版機能のオプトインが必要な可能性があります gcloud beta dataflow flex-template build $TEMPLATE_PATH \ --image "$TEMPLATE_IMAGE" \ --sdk-language "PYTHON" \ --metadata-file "metadata.json"

3. ファイルが Cloud Storage のテンプレート用の場所にアップロードされていることを確認します。

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 Flex テンプレートを作成してステージングする

タスク 3: UI でテンプレートを実行する

このタスクを完了するには、以下の手順に沿って操作します。

1. GCP コンソールの Cloud Dataflow ページに移動します。

2. [テンプレートからジョブを作成] をクリックします。

3. [ジョブ名] フィールドに有効なジョブ名を入力します。

4. [Cloud Dataflow テンプレート] プルダウン メニューから [カスタム テンプレート] を選択します。

5. テンプレートの Cloud Storage パスのフィールドに、テンプレート ファイルへの Cloud Storage パスを入力します。

6. [必須パラメータ] に適切な項目を入力します。 a. [入力ファイルのパス] に、「」を入力します。

   b. [出力ファイルの場所] に、「」を入力します。

   c. [BigQuery 出力テーブル] に「」を入力します。

7. [ジョブを実行] をクリックします。

注: ステージング バケットを指定する必要はありません。Dataflow は、プロジェクト番号を使用して のようにプライベートなステージング バケットをプロジェクト内に作成します。

8. Compute Engine コンソールを確認すると、コンテナを実行して指定のパラメータでパイプラインを開始するために一時的なランチャー VM が作成されています。

タスク 4: gcloud を使用してテンプレートを実行する

1. Dataflow テンプレートを使用する利点の一つは、開発環境以外のさまざまなコンテキストで実行できることです。それを確認するために、gcloud を使用してコマンドラインで Dataflow テンプレートを実行します。

2. このタスクを完了するには、以下のコマンドを、適宜パラメータを変更してターミナルで実行します。

export PROJECT_ID=$(gcloud config get-value project) export REGION={{{project_0.startup_script.lab_region|Region}}} export JOB_NAME=mytemplate-$(date +%Y%m%H%M$S) export TEMPLATE_LOC=gs://${PROJECT_ID}/templates/mytemplate.json export INPUT_PATH=gs://${PROJECT_ID}/events.json export OUTPUT_PATH=gs://${PROJECT_ID}-coldline/template_output/ export BQ_TABLE=${PROJECT_ID}:logs.logs_filtered gcloud beta dataflow flex-template run ${JOB_NAME} \ --region=$REGION \ --template-file-gcs-location ${TEMPLATE_LOC} \ --parameters "inputPath=${INPUT_PATH},outputPath=${OUTPUT_PATH},tableName=${BQ_TABLE}"

3. パイプラインの実行が正常に完了することを確認します。

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 UI と gcloud を使用してテンプレートを実行する

ラボを終了する

ラボが完了したら、[ラボを終了] をクリックします。ラボで使用したリソースが Google Cloud Skills Boost から削除され、アカウントの情報も消去されます。

ラボの評価を求めるダイアログが表示されたら、星の数を選択してコメントを入力し、[送信] をクリックします。

星の数は、それぞれ次の評価を表します。

• 星 1 つ = 非常に不満
• 星 2 つ = 不満
• 星 3 つ = どちらともいえない
• 星 4 つ = 満足
• 星 5 つ = 非常に満足

フィードバックを送信しない場合は、ダイアログ ボックスを閉じてください。

フィードバックやご提案の送信、修正が必要な箇所をご報告いただく際は、[サポート] タブをご利用ください。

Copyright 2026 Google LLC All rights reserved. Google および Google のロゴは、Google LLC の商標です。その他すべての社名および製品名は、それぞれ該当する企業の商標である可能性があります。