Cloud Dataflow ストリーミング パイプラインの管理性が向上
2016年8月23日火曜日
* この投稿は米国時間 8 月 11 日、Software Engineer である Matt Lang によって投稿されたもの（投稿はこちら）の抄訳です。
Google Cloud Dataflow のストリーミング ジョブを停止するための新オプション Drain を皆さんにご紹介できることを、私たちはうれしく思います。
Drain を使ってジョブを停止すると、Dataflow は、すべてのインフライト データが処理されるのを待ってからジョブを穏便にシャットダウンします。このオプションの追加により、Cloud Dataflow サービスで実行されるストリーミング パイプラインの管理がより柔軟になり、既存のオプションとはまったく異なる新しい処理セマンティクスが提供されます。
以下では、Dataflow のストリーミング パイプラインに変更を加える際に、Drain オプションをはじめ、Cancel や Update をどのように使い分けたらよいのかを説明します。
本番システムはどれもそうですが、本番パイプラインを管理するにあたっては、定期的な（場合によっては継続的な）アップデートが必要になります。
たとえば、稼働中のゲーム アナリティクス パイプラインを需要に応じてスケーリングするときに、レコード間の時間的な差がどの程度短ければセッションと判断するかに関して調整を加えたいものとします。バッチ システムの場合は実行の間にシステムに変更を加えて新コードをデプロイするだけなので話は簡単ですが、ストリーミングではこの種の変更は複雑になります。
Cloud Dataflow では、こうした作業を少しでも楽にするために、パイプライン コードの更新方法を選択できるようになっています。
Cloud Dataflow のストリーミング パイプラインを更新するときは、まず 2 つの選択肢からどちらかを選びます。Dataflow の Update 機能を使って既存のパイプラインを新バージョンに置き換えるか、手動で既存パイプラインを停止して交換し、新しいパイプラインを起動するかです。
実行中の Dataflow ジョブは、多くの場合、インフライト データを持っています。インフライト データは、必ず入力から読み込まれ、バッファリングされているものの、まだシンクに送り出されていないデータです。
パイプラインは、たとえば Dataflow の Windowing や Triggers 機能を用いて、イベント時間に基づいてデータをウィンドウにまとめることがあります。また、バッファリング済みデータを抱えた “オープン” ウィンドウが、パイプラインに多数残っていることもあります。バッファリングされているのは、ウィンドウのイベント トリガを遅らせるイベント時間の透かしの設定待ちをしているデータです。
Update を使用すると、Dataflow の exactly-once（正確に 1 回）保証 1 を維持しながら、既存のパイプラインをその場で新バージョンに置き換えます。インフライト データ（既存パイプラインがバッファリングしたデータ）はすべて残され、新パイプラインで処理されます。2 つのジョブの間でデータの重複処理は発生しません。
Cloud Dataflow ジョブの Update は簡単です。そのジョブの名前を --jobNameフラグに指定し、--update フラグ付きで新バージョンのジョブを実行するだけです。他のストリーミング システムとは異なり（たとえばApache Spark の場合は、新コードでは旧パイプラインのチェック ポイントを使えません）、あらかじめ何かを構成したり、手動でジャーナリングを追加したり、パイプラインのアップデートのために特別なコードを組み立てたりする必要はありません。Cloud Dataflow サービス上で実行されるすべてのストリーミング パイプラインは、まったく準備することなくアップデートできます。
Update 機能は、インフライト データを残して新しいパイプラインで処理を再開するようにします。そのため、旧パイプラインの永続ストアを新パイプラインで再利用します。
したがって、新旧 2 つのパイプラインは互換要件を満たしていなければなりません。たとえば、アップデート中に Coders や Side Inputs を変更することはできません。新パイプラインでパイプライン ステップの名前を変えたり、ステップを取り除いたりした場合は、新旧の名前のマッピングを提供する必要があります。
Dataflow サービスは、新しいジョブの互換性チェックを行い、両パイプラインの互換性を保証します2。新しいジョブが非互換であれば、そのジョブは起動されず、古いジョブが実行を続けます。
Update は、パイプラインで使用されるワーカーの数の上限を手動でスケーリングするときにも使えます。Update の詳細は、Cloud Dataflow サービスのドキュメントを参照してください。
新旧のパイプラインが非互換になるような変更が必要になることがあります。その場合、Update の互換チェックはエラーを起こします。そのため、既存パイプラインから非互換の新パイプラインに切り替えるときは、既存の Dataflow ジョブを停止し、その代わりとして新パイプラインを起動しなければなりません。
このようなジョブの停止に対応するため、Dataflow は新たに Cancel と Drain の 2 種類のオプションを提供しています。
Cloud Dataflow Monitoring Interface の新しい “Stop Job” ダイアログでは、Cancel と Drain のいずれかを選択できます。パイプラインの Drain は、gcloud alpha dataflow jobs drain
パイプラインを Cancel に設定すると、ジョブは終了し、ジョブ関連のすべてのリソース（Google Compute Engine 仮想マシン、永続ディスクなど）は開放されます。そして、パイプラインは直ちに終了し、インフライト データやバッファリング済みデータはすべて失われます。
Cancel のときに新パイプラインで実行を再開しても、ジョブで処理されたデータの保証という点では最も弱いものとなります。ジョブ間でデータの重複処理が発生しないことだけは保証されますが（たとえば、入力ソースが exactly-once セマンティクスを提供していない場合）、インフライト データの消失は起こります。
これに対して Drain では、直ちにジョブが停止することはありません。ジョブの停止のために Drain を使うと、ジョブは入力ソースの読み出しを止めるものの、インフライトおよびバッファリング済みのデータの処理は最後まで行われ、オープン ウィンドウの内容を押し出すトリガはすべて発生します。
ただし、この場合、ウィンドウは不完全なものになることがあります。ドレイン中に閉じられるウィンドウには、ドレイン開始時にすでに入力ソースから読み出され、バッファリングされたデータしか含まれていないのです。
Drain は、Cloud Pub/Sub ソースから取り出されたすべてのメッセージが認識されることを保証し、すべての無制限カスタム ソースの finalize() 呼び出しを保証します。バッファリングされたデータがすべて処理され、ジョブに与えられていたすべてのリソースが開放されたことを Dataflow が検出すると、ジョブは停止します。
このように、Drain は at-least-once（少なくとも 1 回）セマンティクスを提供します。これは、Update の強力な保証と、Cancel の極めて弱い保証の中間点と言えるでしょう。ソースが exactly-once を保証しない場合はデータの重複処理が発生することがありますが、すべてのインフライト データの処理は保証します（インフライト データの扱いについては Update と似ています）。
Google Cloud Dataflow のストリーミング ジョブを停止するための新オプション Drain を皆さんにご紹介できることを、私たちはうれしく思います。
Drain を使ってジョブを停止すると、Dataflow は、すべてのインフライト データが処理されるのを待ってからジョブを穏便にシャットダウンします。このオプションの追加により、Cloud Dataflow サービスで実行されるストリーミング パイプラインの管理がより柔軟になり、既存のオプションとはまったく異なる新しい処理セマンティクスが提供されます。
以下では、Dataflow のストリーミング パイプラインに変更を加える際に、Drain オプションをはじめ、Cancel や Update をどのように使い分けたらよいのかを説明します。
ストリーミング パイプラインの変更
たとえば、稼働中のゲーム アナリティクス パイプラインを需要に応じてスケーリングするときに、レコード間の時間的な差がどの程度短ければセッションと判断するかに関して調整を加えたいものとします。バッチ システムの場合は実行の間にシステムに変更を加えて新コードをデプロイするだけなので話は簡単ですが、ストリーミングではこの種の変更は複雑になります。
Cloud Dataflow では、こうした作業を少しでも楽にするために、パイプライン コードの更新方法を選択できるようになっています。
Cloud Dataflow のストリーミング パイプラインを更新するときは、まず 2 つの選択肢からどちらかを選びます。Dataflow の Update 機能を使って既存のパイプラインを新バージョンに置き換えるか、手動で既存パイプラインを停止して交換し、新しいパイプラインを起動するかです。
Update の使い方
実行中の Dataflow ジョブは、多くの場合、インフライト データを持っています。インフライト データは、必ず入力から読み込まれ、バッファリングされているものの、まだシンクに送り出されていないデータです。
パイプラインは、たとえば Dataflow の Windowing や Triggers 機能を用いて、イベント時間に基づいてデータをウィンドウにまとめることがあります。また、バッファリング済みデータを抱えた “オープン” ウィンドウが、パイプラインに多数残っていることもあります。バッファリングされているのは、ウィンドウのイベント トリガを遅らせるイベント時間の透かしの設定待ちをしているデータです。
Update を使用すると、Dataflow の exactly-once（正確に 1 回）保証 1 を維持しながら、既存のパイプラインをその場で新バージョンに置き換えます。インフライト データ（既存パイプラインがバッファリングしたデータ）はすべて残され、新パイプラインで処理されます。2 つのジョブの間でデータの重複処理は発生しません。
Cloud Dataflow ジョブの Update は簡単です。そのジョブの名前を --jobNameフラグに指定し、--update フラグ付きで新バージョンのジョブを実行するだけです。他のストリーミング システムとは異なり（たとえばApache Spark の場合は、新コードでは旧パイプラインのチェック ポイントを使えません）、あらかじめ何かを構成したり、手動でジャーナリングを追加したり、パイプラインのアップデートのために特別なコードを組み立てたりする必要はありません。Cloud Dataflow サービス上で実行されるすべてのストリーミング パイプラインは、まったく準備することなくアップデートできます。
Update 機能は、インフライト データを残して新しいパイプラインで処理を再開するようにします。そのため、旧パイプラインの永続ストアを新パイプラインで再利用します。
したがって、新旧 2 つのパイプラインは互換要件を満たしていなければなりません。たとえば、アップデート中に Coders や Side Inputs を変更することはできません。新パイプラインでパイプライン ステップの名前を変えたり、ステップを取り除いたりした場合は、新旧の名前のマッピングを提供する必要があります。
Dataflow サービスは、新しいジョブの互換性チェックを行い、両パイプラインの互換性を保証します2。新しいジョブが非互換であれば、そのジョブは起動されず、古いジョブが実行を続けます。
Update は、パイプラインで使用されるワーカーの数の上限を手動でスケーリングするときにも使えます。Update の詳細は、Cloud Dataflow サービスのドキュメントを参照してください。
手動での交換
新旧のパイプラインが非互換になるような変更が必要になることがあります。その場合、Update の互換チェックはエラーを起こします。そのため、既存パイプラインから非互換の新パイプラインに切り替えるときは、既存の Dataflow ジョブを停止し、その代わりとして新パイプラインを起動しなければなりません。
このようなジョブの停止に対応するため、Dataflow は新たに Cancel と Drain の 2 種類のオプションを提供しています。
Cloud Dataflow Monitoring Interface の新しい “Stop Job” ダイアログでは、Cancel と Drain のいずれかを選択できます。パイプラインの Drain は、gcloud alpha dataflow jobs drain
パイプラインを Cancel に設定すると、ジョブは終了し、ジョブ関連のすべてのリソース（Google Compute Engine 仮想マシン、永続ディスクなど）は開放されます。そして、パイプラインは直ちに終了し、インフライト データやバッファリング済みデータはすべて失われます。
Cancel のときに新パイプラインで実行を再開しても、ジョブで処理されたデータの保証という点では最も弱いものとなります。ジョブ間でデータの重複処理が発生しないことだけは保証されますが（たとえば、入力ソースが exactly-once セマンティクスを提供していない場合）、インフライト データの消失は起こります。
これに対して Drain では、直ちにジョブが停止することはありません。ジョブの停止のために Drain を使うと、ジョブは入力ソースの読み出しを止めるものの、インフライトおよびバッファリング済みのデータの処理は最後まで行われ、オープン ウィンドウの内容を押し出すトリガはすべて発生します。
ただし、この場合、ウィンドウは不完全なものになることがあります。ドレイン中に閉じられるウィンドウには、ドレイン開始時にすでに入力ソースから読み出され、バッファリングされたデータしか含まれていないのです。
Drain は、Cloud Pub/Sub ソースから取り出されたすべてのメッセージが認識されることを保証し、すべての無制限カスタム ソースの finalize() 呼び出しを保証します。バッファリングされたデータがすべて処理され、ジョブに与えられていたすべてのリソースが開放されたことを Dataflow が検出すると、ジョブは停止します。
このように、Drain は at-least-once（少なくとも 1 回）セマンティクスを提供します。これは、Update の強力な保証と、Cancel の極めて弱い保証の中間点と言えるでしょう。ソースが exactly-once を保証しない場合はデータの重複処理が発生することがありますが、すべてのインフライト データの処理は保証します（インフライト データの扱いについては Update と似ています）。
| 方法 | 処理セマンティクス |
| その場で Update | exactly-once |
| Drainして交換 | At-least-once3 |
| Cancel して交換 | なし |
詳細は Cloud Dataflow ドキュメントのパイプラインの停止に関するページを参照してください。
これだけのツールが揃えば、処理セマンティクスの要件に合わせてパイプライン コードを更新するのは簡単です。パイプラインがメンテナンスの重荷になることを恐れる必要はなくなるでしょう。
- Posted by Matt Lang, Software Engineer
これだけのツールが揃えば、処理セマンティクスの要件に合わせてパイプライン コードを更新するのは簡単です。パイプラインがメンテナンスの重荷になることを恐れる必要はなくなるでしょう。
1 Google Cloud Dataflow マネージド サービスは、ストリーミング パイプラインに対して exactly-once セマンティクスを保証します。つまり、データを重複して送ることがあるソースを使っても、Dataflow パイプラインはすべてのユニークなデータを正確に 1 回必ず処理します。
3 カスタム ソースが exactly-once デリバリを保証し、ソース サイド バッファリングを提供する場合は、“Drain して交換” のときでも exactly-once セマンティクスを提供できます。
- Posted by Matt Lang, Software Engineer
Nice post i see ever. Information is so reliability. Thanks for share us.
返信削除Microsoft Office365