G-gen の杉村です。 Google Cloud (旧称 GCP) には Cloud Workflows という簡易的なワークフローツールがあります。今回の記事は Cloud Workflows を使った簡易的なデータパイプラインの構築方法をご紹介します。

Cloud Workflows とは

Cloud Workflows は Google Cloud (旧称 GCP) で提供されるワークフローツールです。

事前に定義したワークフロー (いわゆるジョブネット) に沿って、自動化ジョブを実行できます。ワークフローからは HTTP リクエスト (GET/POST) を発行したり、各種 Google Cloud API を呼び出すことができます。

ワークフローではジョブを「順次」「分岐」「繰り返し」を利用して実行可能です。

フルマネージドサービスなのでインフラの管理・運用は一切不要であり、安価で無料枠も大きいのが特徴です。

ただしワークフローを GUI で編集することは現状できず、フローを yaml 形式 で記述します (簡易的な可視化機能はあります) 。

また、ワークフローは Cloud Scheduler と組み合わせて定期的な自動実行が可能です。

この記事で作るもの

概要

当記事では Cloud Workflow を使って「 BigQuery のデータに対して SQL を実行して ELT 処理を行うワークフロー」を構築してみます。

当記事で紹介する手法では BigQuery Data Transfer API を利用するため「 Amazon S3 からデータを BigQuery に取り込み。その後、 ELT 処理を実行」というパイプラインも実現可能です。

SQL 実行方法 1 - BigQuery API Connector

Cloud Workflows には Google Cloud の API を簡単に呼び出すための各種コネクタが存在します。

BigQuery API Connector もその一つで、簡単に BigQuery に対して SQL を実行できます。

ただし、このコネクタを使うとワークフローの yaml に直接 SQL を記述する必要があり、保守性が悪いため、今回はこの方法は 採りません 。

SQL 実行方法 2 - Scheduled Query

BigQuery の Scheduled Query (スケジュールされたクエリ) 機能では、 SQL を保存しておいて後から呼び出すことができます。

Scheduled Query は本来、定期的に SQL を自動実行するための BigQuery 備え付けの機能です。しかし呼び出されたときだけ SQL を実行する「オンデマンド」としての保存もできるため、今回はこれを呼び出す形を採ります。

Scheduled Query の呼び出し方

Scheduled Query は BigQuery Data Transfer API 経由で呼び出すことができます。

Cloud Workflow には BigQuery Data Transfer API をノーコードで呼び出せる BigQuery Data Transfer API Connector も存在するのですが 2022 年 6 月現在で Preview 段階です。

本番利用を想定して、当記事ではワークフローから Cloud Functions の関数を呼び出し、その関数から BigQuery Data Transfer API をコールする形でデータパイプラインを実現していきます。

なお BigQuery Data Transfer API では他にも Amazon S3 から BigQuery へデータを転送するジョブ等も呼び出せるため、当記事のワークフローで「 Amazon S3 からデータを BigQuery に取り込み。その後、 ELT 処理を実行 」といった処理も可能です。

ワークフローの作成

フロー図

この記事では、以下の図のようなワークフローを構築します。

以下の図では実行されるジョブが query_01 の一つだけですが、数珠つなぎにジョブを繋げることで、クエリの完了を待って次のクエリを実行することが可能です。

Cloud Functions 関数

今回作成する Cloud Functions 関数については当記事では詳細は紹介しませんが、以下のような関数を想定しています。

No	関数名	処理内容	戻り値
1	execute-sql	BigQuery Data Transfer API 経由で、与えられたリソース名の Scheduled Query を実行する	実行したジョブ名例: projects/1234567890/locations/asia-northeast1/transferConfigs/11111111-0000-1111-1111-111111111111/runs/99999999-0000-9999-9999-999999999999
2	check-status	BigQuery Data Transfer API 経由で、与えられたジョブ名のジョブのステータスを確認する	ジョブのステータス例: SUCEEDED / RUNNING / PENDING / FAILED 等

当記事ではこれらの関数が HTTP トリガ関数としてデプロイ済みである前提で解説します。

以下の資料をご参照ください。

Google CloudのCloud Functionsを徹底解説！(基本編)
Package google-cloud-bigquery-datatransfer (3.6.2) ※ Python 用 Cloud SDK のリファレンス

yaml ファイル

以下のような yaml ファイルを作成します。

main:
  steps:
    - initialize:
        assign:
          - query_01_name: "projects/1234567890/locations/asia-northeast1/transferConfigs/11111111-0000-1111-1111-111111111111"
          - query_01_timeout: 300
        next: query_01
    - query_01:
        steps:
          - execute_01:
              call: http.post
              args:
                url: https://asia-northeast1-your-project-name.cloudfunctions.net/execute-sql
                body:
                  scheduled_query_name: ${query_01_name}
                auth:
                  type: OIDC
              result: run_name
          - check_01:
              steps:
                - setup_01:
                    assign:
                      - elapsed_time: 0
                      - timeout: ${query_01_timeout}
                - check_status_01:
                    call: http.post
                    args:
                      url: https://asia-northeast1-your-project-name.cloudfunctions.net/check-status
                      body:
                        run_name: ${run_name.body}
                      auth:
                        type: OIDC
                    result: status
                - check_if_complete_01:
                    switch:
                      - condition: ${status.body == "SUCCEEDED"}
                        next: finalize
                      - condition: ${elapsed_time >= timeout}
                        raise: "Job timed out"
                      - condition: ${status.body == "RUNNING" OR status.body == "PENDING"}
                        assign:
                          - elapsed_time: ${elapsed_time+10}
                        next: wait_01
                      - condition: True
                        raise: "Job error"
                - wait_01:
                    call: sys.sleep
                    args:
                      seconds: 10
                    next: check_status_01
    - finalize:
        return: "SUCEEDED"

7 行目の以下の行は、実際の Scheduled Query として保存されたオンデマンドクエリのリソース名で置き換えてください。

query_01_name: "projects/1234567890/locations/asia-northeast1/transferConfigs/11111111-0000-1111-1111-111111111111"

15 行目および 31 行目の以下の 2 行は Cloud Fucntions 関数 (HTTP トリガ) の呼び出し URL ですので、実際に構築した Cloud Functions のものに置き換えてください。

url: https://asia-northeast1-your-project-name.cloudfunctions.net/execute-sql
url: https://asia-northeast1-your-project-name.cloudfunctions.net/check-status

実行の流れ

冒頭のフロー図に示したとおり、これにより以下のワークフローが定義されます。

execute-sql 関数が実行されジョブが投入される
check-status 関数が実行されジョブのステータスが確認される
ステータスにより分岐 (以下の順で評価)
- ステータスが SUCEEDED であればジョブが正常終了
- 設定したタイムアウト秒数を超えていればワークフローが異常終了
- ステータスが PENDING や RUNNING であれば 10 秒待機
- 10 秒間待機
check-status 関数の実行まで戻る

ワークフローのデプロイ

yaml ファイルができたら、以下のコマンドを実行します。

gcloud workflows deploy example-workflow \
--location=asia-northeast1 \
--source=example-workflow.yaml \
--service-account=hogehoge@hogehoge.gserviceaccount.com

4 行目の --service-account=hogehoge@hogehoge.gserviceaccount.com は、 Cloud Functions を起動する権限のあるサービスアカウントに置き換えてください。

ワークフローの実行

以下のコマンドにより、デプロイしたワークフローが実行可能です。

gcloud workflows run example-workflow --location=asia-northeast1 --call-log-level=log-errors-only

ログレベルの設定など詳細はドキュメントをご確認ください。このコマンドでは、各ステップの実行履歴はログ出力されず、エラーが起きた場合のみログが Cloud Logging に出力されます。

なお、もちろん Cloud Workflows のコンソール画面からもワークフローのデプロイや実行が可能なほか、実行状態を確認することができます。

gcloud コマンドで実行した場合はワークフロー全体が完了するまでプロンプトが戻らないので、適宜コンソールで状態を確認したり、別ターミナルからステータスを確認するコマンドを実行します。

留意事項

データパイプラインをはじめ自動化ジョブではエラー時の検知やリトライを検討する必要があります。

当記事でご紹介したワークフローはあくまで簡易的なものです。実際の業務では以下を考慮してください。

冪等性 (リトライ性)
エラーの検知

ワークフローが「コケた」際に、シンプルにワークフローを再実行すれば良いように、冪等性 (何度実行しても同じ結果になる性質) が担保されるように変換処理を設計します。

またエラー検知に関しては、ワークフローのエラーを Cloud Logging に出力させて Cloud Monitoring で検知してメールや Slack へ通知する等が考えられます。

当記事のワークフローでは関数のエラー/例外をワークフロー内でハンドリングしていません (関数がコケても戻り値が返ると次のステップが実行されてしまう) 。 Cloud Workflows にはエラー/例外をハンドリングする仕組みもありますので、適宜利用します。