バッチ処理のスケジューリングパターン
この記事はこの記事は Google Cloud Japan Customer Engineer Advent Calendar 2019 の 12日目の記事です。
はじめに
Google Cloud Platform (GCP) でバッチ処理を起動するための以下のパターンについてご紹介したいと思います。以下、８パターンあげてみました。とはいえ、最後の3つは GCP のバッチスケジューリングという観点からは少し外れますが、バッチの起動時に使われるということでご容赦を。
- Cloud Scheduler : フルマネージドな cron ジョブスケジューラです。フルマネージドという点が非常に大きなメリットであり、多くの処理を自動化し実行することが可能です。
- Google App Engine cron サービス : HTTP GET を利用して、特定の URLを呼び出します。Google AppEngine( GAE )の処理を呼び出すには最適な選択肢となります。
- Cloud Composer : Apache Airflow をベースに構築されたワークフロー オーケストレーション サービスです。Cloud Composer はスケジューリングだけでなく、バッチ処理のパイプラインの作成やモニタリングを行うことが可能です。
- Cloud Data Fusion : 先日の Google Cloud Next ’19 UK で GA ( general availability ) が発表されました。Cloud Data Fusion は データ統合サービスです。UIで用意されたパイプライン作成機能や、あらかじめ用意されたコネクタ類を利用し、ノンコーディングでパイプラインの作成が可能です。
- BigQuery のスケジュールクエリ : BigQuery に用意されたスケジューラを利用して、定期的にクエリを実行することが可能です。DML と DDL を含めることができます。また、パラメータを渡して処理を実行することも可能です。
- イベントトリガー with Cloud Function : Cloud Function を利用し、HTTP のリクエスト 、 Google Cloud Storage ( GCS )のオブジェクトの追加や変更、Pub/Sub のメッセージなどのイベントをトリガに処理を実行することができます。
- イベントトリガー with Cloud Pub/Sub : Cloud Pub/Sub を利用し、GCSのオブジェクトの追加や変更、Stackdriver Logging のメッセージをトリガに Cloud Pub/Sub のトピックにメッセージを Publish し、処理を実行することができます。
- Something on GCE : 何かしらのスケジューラを Google Compute Engine ( GCE ) 上で起動させることが可能です。
以降、それぞれの処理について見ていきます。
Cloud Scheduler
バッチ処理のスケジュールをしたいと思った時に、一番最初に思いつくサービスではないでしょうか。何故ならば、名前がそのまま Cloud Scheduler！Cloud Scheduler は、GAE に依存して動作し、後述のAppEngine cron サービスと同等のスケジュールの設定が可能です。しかし、そこであえて、Cloud Scheduler を使うメリットとしては以下が挙げられます。
・AppEngine のアプリケーション以外に、Pub/Sub のトピック、HTTP/S のエンドポイントをターゲットに指定し、スケジューラの実行が可能です。
・gcloud コマンド、もしくは GCP のコンソールで設定が 可能です。
GCP のサービスの処理を起動する場合には、Pub/Sub を経由して処理を実行する方がおすすめです。以下、Dataflow の処理を定期的に実行するフローとなります。
下記のように、Pub/Sub を利用しない場合には、Cloud Function では、HTTP エンドポイントで Cloud Scheduler からのリクエストを受け取ることになります。その場合には、目的の Cloud Scheduler から呼び出されたことを確認するために、何かしらの認証を行うことが必要です。また、不正なリクエストを Cloud Function 内で不正と判断できたとしても、Cloud Function の料金がかかってしまうことになります。外部のサービスをSchedulerから起動する場合には、こちらをご参考ください。( Cloud Run はcloud run invoker 役割を設定することにより、IAM で実行可否の制御が可能です。）
Google App Engine cron サービス
Google App Engine cron サービスを利用すると、GAE のサービスに対して定期的にリクエストを送ることが可能となります。cron.yaml を設定し、どの URL に対してリクエストを送るか指定します。GAE でアプリケーションが完結している場合にはもっともシンプルにバッチ処理を実現できます。デプロイは GAE のアプリケーションと同様に以下のコマンドを実行して行います。
gcloud app deploy cron.yamlCloud Composer
Cloud Composer は Apache Airflow をベースとしたワークフローオーケストレーションです。先述の Cloud Scheduler は単発のバッチ処理を実行する場合には、シンプルで使いやすいのですが、この Composer は複数のバッチを組み合わせてバッチ処理のワークフローを構築することが可能です。
ワークフローは Dag と呼ばれるファイルで、python を用いて記述されます。また、Operator と呼ばれるライブラリを用いることによって、Dag ファイルの中に定義を行い、処理を記述することが可能です。
例として、次のワークフローを考えてみましょう。
- ロード先の BigQuery に特定のデータセットが存在しなければ作成する。
- GCS からファイルをエクスポートして、BigQuery にロードしたい。
- データのロード後、サマリーデータを別のテーブルに格納する。
上記の処理は、以下の Operator でそれぞれ実行ができます。
・特定のコマンド実行 : BashOperator
・GCS → BigQuery へのロード : GoogleCloudStorageToBigQueryOperator
・BigQuery の処理実行 : BigQueryOperator
GCS → BigQuery のロード時に複雑な ETL 処理が必要な場合には、Dataflow で処理を記述し、dataflow_operatorで処理を呼び出すことが可能となります。
dataflow_operator を利用することで、Dataflow の処理の完了まで Polling し待つことができるため、Dataflow の処理の後に、次の処理を実行するというような処理を簡単に実装できます。
Cloud Data Fusion
Cloud Data Fusion は CDAP を利用したマネージドサービスになります。Cloud Composer との大きな違いは、何と言っても GUI で ETL 処理を作成することができることじゃないでしょうか。
こちらの画面はスタジオと呼ばれるデータ連携処理の作成画面です。スタジオでは最初から含まれるリソースの他に、Hub と呼ばれるエリアから、コネクタやパイプラインを取得し、カスタマイズすることで処理を実行することが可能です。先日の Google Cloud Next’19 UK で GA となったばかりですが、今後も機能が拡張されていくことを期待しています。スケジューリングも GUI からきめ細かく設定できる点がうれしいところだと思います。
BigQuery のスケジュールクエリ
BigQuery にはスケジューリング機能があります。このスケジューラは Data Transfer (*1) サービスを利用しています。スケジュールクエリ機能を利用して、以下のことを行うことができます。
- 毎時、毎日、毎週などのスケジュールを設定し、定期的に BigQuery でクエリを実行可能です。( DML 、DDL も可能 )
- スケジュールにオンデマンドを指定し、コンソール、もしくは API で処理を実行可能です。
- 実行結果を特定のテーブルに書き込みが可能です。また、書き込みの際に、既にテーブルが存在する場合には、追加書き込み、もしくは上書きの選択をあらかじめ設定しておくことが可能です。
- 処理の終了時に、メールの通知、Pub/Sub トピックへの通知が可能です。そのため、BigQuery のクエリを連動されたワークフローであれば、以下の構成も可能となります。
- Backfill の機能を利用して、過去に遡って処理の実行が可能です。
- パラメータを指定し、検索条件を変更したり、書き込み先テーブルのパーティションなどを指定することが可能です。
BigQuery のスケジュールクエリを利用することで、BigQuery で行われる処理においてはかなり幅広いユースケースをカバーできそうです。
*1 : 本ブログでは Data Transafer サービスには触れませんでしたが、Data Transfer サービスを利用することで、データの転送を定期的に行うことが可能となります。
イベントトリガー
イベントトリガーは、バッチスケジューリングとはちょっとずれた話題になります。ただ、バッチ処理の起動のトリガーとしては使われることがあると思いますので、ここで取り上げてみました。
・Cloud Function : イベントによって処理の実行が可能です。Firebase やFirestore との連携もありますが、バッチ処理という観点では、HTTP トリガー、Cloud Pub/Sub トリガー、Cloud Storage トリガーが使えることを覚えておくといいと思います。
例えば、オンプレミスで処理したデータを gsutil ( Cloud Storage のコマンドラインツール ) でアップロードしたファイルを、アップロード直後に処理をする場合には、Cloud Storage トリガー利用できます。
また、Cloud Scheduler で記載したように、メッセージングによって処理を起動する場合には、Cloud Pub/Sub トリガーを利用することができます。
・Cloud Pub/Sub : Stackdriver 、Cloud Storage などと連携して、イベントを取得することができます。Cloud Function より汎用的に必要な処理を実行することができます。例えば、Cloud Function では、タイムアウトの制限があるため、長時間のバッチ処理を行うことはできませんが、Pub / Subのメッセージを GCE でサブスクライブし実行すると、長時間のバッチ処理が可能となります。処理を行う GCE のインスタンスは、Cloud Schduler を利用して、必要なタイミングで起動させることができます。
Something on GCE
ユーザーの好みに合わせたツールを入れることが可能です。各ツールがGCP 上で動作がサポートされているかについては、各ベンダーにお問い合わせください。これまで紹介してきたマネージドのスケジューラはマネージドサービスであり、SLA を維持するよう GCP 内部で可用性を保たれています。GCE 上で構築される場合には、スケジューラが単一障害点になり得るため、注意して設計することをお勧めします。
どのパターンを選択するか？
ここまで、ずらーっとバッチのスケジューリングパターンをあげてきました。同じ処理を行うにしても、複数の選択肢があって悩む場合があると思います。その場合には、まずはユースケースを整理してみましょう。そもそもバッチ処理が最適なのか、ストリーミング処理が最適なのかを検討するとまた別の実装方法が見えてくるかもしれません。
また、バッチ処理全体の規模で考えてみてはいかがでしょうか？
バッチ処理全体の規模が大きい場合 : 多様なバッチ処理が考えられます。Dataflowを利用したり、BigQuery の処理があったりする場合に、それぞれのコンポーネントでスケジューリングを行うと管理が大変になります。そのため、Cloud Composer などで一元管理を行うと、良いかと思います。またイベントを利用して、複雑なワークフローを記述することもできますが、ワークフローがソースコードに記述されてしまうと可読性が低下してしまいますので、運用の観点においてはワークフローを切り出した方が良いかと思います。
バッチ処理の全体の規模が小さい場合：小さなプロジェクトで一部の処理にバッチ処理を組み込みたいケースがあると思います。その場合には、Cloud Composer で管理すると管理コストのオーバーヘッドや Cloud Composer そのもののコストが負担になる可能性があります。そのため、各処理に近いところで処理を行うことが良さそうです。BigQuery で行われる処理であれば BigQuery のスケジュールクエリを利用することが簡単です。
また GCP の各処理は API から実行することが可能です。そのため、小さく始めたプロジェクトであっても、プロジェクトの規模が大きくなってきた場合には、Composer などに移行し、API 経由で処理を実行することが可能です。
さいごに
GCP にはスケジューリング実行のための多くの機能が備わっています。これは、開発者が本来の目的である開発に集中してほしいというメッセージであると思います。ぜひみなさんのプロジェクトにおいて最適な方法を探してみてください！
明日は、Yuki-san( iwanary ) による 「アンチパターンから学ぶ！Cloud Spanner あるあるクイズ 」です。とても注目度の高い Spanner の記事で個人的にもすごく楽しみにしています！
それでは Enjoy hacking and the holidays !
