Pub/Sub を使用したストリーミング

このページでは、Dataflow と Pub/Sub の統合の概要について説明します。また、Dataflow ランナーで Pub/Sub I/O コネクタを実装する際に利用できる最適化についても説明します。Pub/Sub は、スケーラブルで耐久性のあるイベントの取り込みおよび配信システムです。ウィンドウとバッファリングを使用する場合、Dataflow は、Pub/Sub のスケーラブルな「最低 1 回」配信モデルに、メッセージ重複排除と「1 回限り」「正しい順序」の処理機能を追加します。Dataflow を使用するには、Apache Beam SDK でパイプラインを記述し、Dataflow サービスでパイプラインコードを実行します。

開始する前に、Apache Beam とストリーミングパイプラインの基本的なコンセプトをご確認ください。詳細については、次のリソースをご覧ください。

PCollection、トリガー、ウィンドウ、ウォーターマークなどの Apache Beam のコンセプトの概要
ラムダ以降: Dataflow における「1 回限り」の処理のパート 1 とパート 3: ソースとシンク
ストリーミング: バッチを超えた世界 101、102
Apache Beam プログラミングガイド

警告: 複数のパイプラインに対して 1 つの Pub/Sub サブスクリプションを使用しないでください。1 つのサブスクリプションのデータは、そのデータを消費するエンティティによって非決定論的に分割されます。したがって、1 つのサブスクリプションを読み取る 2 つのパイプラインがある場合、各パイプラインはデータの一部を非決定論的な方法で受け取ることになります。それにより、メッセージの重複、ウォーターマークのラグ、効率の悪い自動スケーリングが発生する可能性があります。パイプラインごとに別のサブスクリプションを作成してください。また、Pub/Sub トピックから読み取る場合は、パイプラインごとに個別のサブスクリプションが自動的に作成されます。

Pub/Sub を使用したストリーミングパイプラインの構築

Dataflow と Pub/Sub の統合の利点を活用するには、次のいずれかの方法でストリーミングパイプラインを構築します。

ストリーミング単語抽出（Java）やストリーミング単語数カウント（Python）など、Apache Beam GitHub リポジトリにある既存のストリーミングパイプラインのサンプルコードを使用します。
Apache Beam API リファレンス（Java または Python）を使用して、新しいパイプラインを記述します。
Google が提供する Dataflow テンプレートと、それに対応する Java のテンプレートソースコードを使用します。

Google は、UI ベースで Pub/Sub ストリーム処理パイプラインを開始するための Dataflow テンプレートを提供しています。Java を使用する場合は、こうしたテンプレートのソースコードを使用して、カスタムパイプラインを作成することもできます。

次のストリーミングテンプレートは、Pub/Sub データを別の宛先にエクスポートします。
次のバッチテンプレートは、Pub/Sub トピックにデータのストリームをインポートします。
- Cloud Storage Text to Pub/Sub (Batch)
Pub/Sub のクイックスタート: Dataflow によるストリーム処理に従って、単純なパイプラインを実行します。

Pub/Sub と Dataflow の統合機能

Apache Beam は、Pub/Sub 向けのリファレンス I/O ソース実装（PubsubIO）を提供しています（Java、Python）。これは、Apache Spark ランナー、Apache Flink ランナー、Direct Runner など、Dataflow 以外のランナーによって使用されます。

Dataflow ランナーは PubsubIO の別のプライベート実装を使用します。この実装では、Google Cloud 内部の API とサービスを利用しています。これにより、低レイテンシのウォーターマーク、高精度のウォーターマーク（つまりデータ完全性）、効率的な重複排除という主要な 3 つの利点が得られます。

Dataflow ランナーの PubsubIO の実装により、メッセージが永続ストレージ（Shuffle またはシンク）に書き込まれると自動的に確認応答します。したがって、一部のコンポーネントがクラッシュした場合や接続が失われた場合で、Dataflow によってデータの損失がないことが保証される場合にのみ、メッセージが確認応答されます。

低レイテンシのウォーターマーク

Dataflow は Pub/Sub のプライベート API にアクセスできます。これにより、Cloud Monitoring で使用可能なレイテンシよりも低いレイテンシで、サブスクリプション内の最も古い未承認メッセージの経過時間を得ることができます。比較のため、Cloud Monitoring で使用可能な Pub/Sub バックログ指標は通常 2～3 分の遅延が発生しますが、Dataflow の指標は 10 秒程度しか遅れません。これにより、Dataflow はパイプラインウォーターマークを先に進めて、ウィンドウ処理された計算結果をより早く出力できます。

高精度のウォーターマーク

イベント時刻で定義されたウィンドウには強力なウォーターマークが必要になるという問題がありますが、この問題も Dataflow と Pub/Sub の統合によってネイティブに解決されます。イベント時間は、Pub/Sub サービス自身によってメッセージに設定された publish_time フィールドではなく、パブリッシャーアプリケーションによって Pub/Sub メッセージの属性として指定されたタイムスタンプです。Pub/Sub では、サービスによって割り当てられた（処理時刻の）タイムスタンプに関してのみバックログ統計が計算されるため、イベント時刻のウォーターマークを推測するには別のメカニズムが必要です。

この問題を解決するため、ユーザーがカスタムのイベントタイムスタンプを使用する場合、Dataflow サービスは第 2 のトラッキングサブスクリプションを作成します。このトラッキングサブスクリプションを使用して、ベースとなるサブスクリプションのバックログ内にあるメッセージのイベント時刻を検査し、イベント時刻のバックログを推測します。詳しくは、StackOverflow の Dataflow で Pub/Sub ウォーターマークを計算する方法に関するページをご覧ください。

効率的な重複排除

「1 回限り」のメッセージ処理を行うには、メッセージの重複排除が必要になります。Dataflow は、Pub/Sub メッセージ ID に基づいてメッセージの重複排除を行います。その結果、すべての処理ロジックにおいて、Pub/Sub メッセージ ID に基づいてメッセージが一意になっていると想定できます。これを達成するための効率的な増分集約メカニズムは、PubsubIO API の中で抽象化されています。

PubsubIO がカスタムメッセージ ID を使用するように構成されている場合、Dataflow は過去 10 分間に確認したすべてのカスタム ID のリストを保持することで、メッセージの重複を除去します。新しいメッセージの ID がこのリストに含まれる場合、メッセージは重複するものとみなされ、破棄されます。