Workflows を使用したデータ パイプラインのオーケストレーション
Google Cloud Japan Team
※この投稿は米国時間 2021 年 6 月 24 日に、Google Cloud blog に投稿されたものの抄訳です。
データ エンジニアリングの問題に対応する際は、実行制御を管理できる機能、実行履歴やログ履歴などの運用情報にアクセスできる機能、障害点からワークフローを再開できる機能が、パイプライン開発全体にとって非常に重要になります。Google Cloud でこれを実現する最も一般的な方法の一つは、Apache Airflow をベースとした Cloud Composer を使用することです。
サーバーレスの方法をお探しの場合は、Workflows を使ってサーバーレス ワークフローを作成して、一連のタスクを定義した順序でまとめて処理できます。このブログでは、Workflows を使って GCP で データフロー パイプラインをオーケストレートする方法をご紹介します。では、さっそく始めましょう。
Workflows とは
ワークフローは、Workflows 構文で記述された一連のステップで構成され、YAML 形式または JSON 形式のいずれでも記述できます。これがワークフローの定義です。Workflows 構文の詳細については、構文リファレンスのページをご覧ください。
ワークフローを作成するとデプロイされ、ワークフローを実行する準備が整います。
実行とは、ワークフローの定義に含まれるロジックを 1 回だけ実行することです。
Workflows を使う理由
費用と効率の観点から見た場合、デベロッパー エクスペリエンスを容易にする鍵となるのが Workflows です。次に例を示します。
従量課金 - Workflows は使用していないときにはゼロにスケールされ、アイドル状態では費用は発生しません。料金はワークフローのステップ数に基づくため、ワークフローが実行された場合にのみ課金されます。
サーバーレス - お客様側での管理が必要な基盤インフラストラクチャがありません。Workflows は「コールド スタート」の影響を受けずに、自動的にスケールアップします。
密接な統合 - Google Cloud プロダクトや HTTP ベースの API（SaaS やプライベート API を含む）全体のイベントを管理します。
実行時間の長いオペレーション - Cloud Functions などのサーバーレス プロダクトを使ってオーケストレーションを設定することもできますが、最大実行時間が 9 分であるため、実行時間の長いワークロードで使用する際には注意が必要です。それに対して Workflows では、最大で 1 年間の実行が可能です。
組み込みのエラー処理 - 指数バックオフによる自動 HTTP 呼び出し再試行、カスタム エラーハンドラ、その他の高度な機能など、ワークフローの例外処理によって、障害発生時にワークフローを復元およびカスタマイズして対応できます。
Workflows を使用したデータ パイプラインのオーケストレーション
Google Cloud のパイプラインのフローと対応するステップを以下に示します。
パイプラインのステップ
このパイプラインでは、入力ファイルは GCS バケットに格納されます。Dataflow ジョブがデータを読み取って BigQuery に格納した後で、Cloud Functions の関数を使ってファイルをアーカイブします。詳細なステップは以下のとおりです。
スケジュール設定された Cloud Scheduler が Workflows ジョブをトリガーします。
Workflows が Dataflow バッチジョブをトリガーして create_dataflow_job タスクを呼び出します。
Dataflow ジョブが取り込み GCS バケットから入力ファイルを読み取ります。
データ変換が完了すると、Dataflow ジョブがデータを BigQuery テーブルに書き込みます。
Dataflow ジョブが完了すると、Workflows が Cloud Functions の関数 call_file_archival_function をトリガーして入力ファイルをアーカイブします。
Cloud Functions の関数が Cloud Storage の ‘processing’ バケットから ‘processed’ バケットに入力ファイルを移動します。
Cloud Scheduler でワークフローをトリガーする
このデータ パイプラインでは、Cloud Scheduler ジョブを利用して Workflows の実行をトリガーします。Cloud Scheduler を利用してワークフローのスケジュールを設定する方法については、こちらをご覧ください。
以下の gcloud コマンドは、認証にサービス アカウントを使用して、ワークフローをトリガーする Cloud Scheduler ジョブを作成します。以下の例では、ワークフローを 5 分ごとに実行するようにスケジュールします。
ワークフローの YAML
1. Dataflow ジョブを開始する
2. 処理の終了後に入力ファイルをアーカイブするための Cloud Functions の関数をトリガーする
ワークフロー ジョブの図
本投稿では、Workflows を使って GCP 上でデータ パイプラインをオーケストレーションする方法をご紹介しました。迅速なスケーリング、すぐに使用できる例外処理、組み込みの認証、Cloud Logging との統合などの機能を備えた Workflows は、サーバーレス実装に最適です。
実際に Workflows を使ってみる
Workflows を使い始めるには、クイック スタート ガイドとチュートリアルをご覧ください。また、Google Cloud APIs とシームレスに統合でき、組み込みの認証機能やエラー処理機能を備えた新しい Workflows コネクタについても、ご確認ください。
-戦略的クラウド エンジニア Preetika Bhateja
-戦略的クラウド エンジニア Kunal Kumar Gupta
