このページでは、Pub/Sub から読み取り、BigQuery に書き込む Dataflow ストリーミング ジョブのパフォーマンス特性について説明します。2 種類のストリーミング パイプラインのベンチマーク テストの結果を示します。
マップのみ（メッセージごとの変換）: ストリーム全体で状態を追跡したり、要素をグループ化したりせずに、メッセージごとの 変換を行うパイプライン。例としては、ETL、フィールド検証、スキーマ マッピングなどがあります。
ウィンドウ集計 （
GroupByKey）: 状態を保持するオペレーションを実行し、キーと時間ウィンドウに基づいてデータをグループ化するパイプライン。例としては、イベントのカウント、合計の計算、ユーザー セッションのレコードの収集などがあります。
ストリーミング データ統合のワークロードのほとんどは、この 2 つのカテゴリに分類されます。パイプラインが同様のパターンに従っている場合は、これらのベンチマークを使用して、パフォーマンスの高い参照構成に対して Dataflow ジョブを評価できます。
テスト方法
ベンチマークは次のリソースを使用して実施されました。
入力負荷が一定の Pub/Sub トピック（プロビジョニング済み）。 メッセージは、 ストリーミング データ ジェネレータ テンプレートを使用して生成されました。
- メッセージ レート: 1 秒あたり約 1,000,000 メッセージ
- 入力負荷: 1 GiB/秒
- メッセージ形式: 固定スキーマのランダムに生成された JSON テキスト
- メッセージ サイズ: メッセージあたり約 1 KiB
標準の standard BigQuery テーブル。
Pub/Sub to BigQuery テンプレートに基づく Dataflow ストリーミング パイプライン。 これらのパイプラインは、必要な最小限の解析とスキーマ マッピングを実行します。カスタム ユーザー定義関数（UDF）は使用されませんでした。
水平スケーリングが安定し、パイプラインが定常状態に達した後、パイプラインを約 1 日間実行し、その結果を収集して分析しました。
Dataflow パイプライン
2 つのパイプライン バリアントをテストしました。
マップのみのパイプライン 。このパイプラインは、 JSON メッセージの簡単なマッピングと変換を行います。このテストでは、 Pub/Sub to BigQuery テンプレート をそのまま使用しました。
- セマンティクス: パイプラインは、 exactly-once モードと at-least-once モードの両方を使用してテストされました。at-least-once 処理では、スループットが向上します。ただし、重複レコードが許容される場合、またはダウンストリーム シンクが重複除去を処理する場合にのみ使用してください。
ウィンドウ集計パイプライン 。このパイプラインは、固定サイズのウィンドウ内の特定の キーでメッセージをグループ化し、集計されたレコードを BigQuery に書き込みます。このテストでは、Pub/Sub to BigQuery テンプレートに基づくカスタム Apache Beam パイプライン を使用しました。
集計ロジック: 重複しない固定の 1 分間のウィンドウごとに、 同じキーを持つメッセージが収集され、1 つの集計 レコードとして BigQuery に書き込まれました。このタイプの集計は、ログ処理でよく使用されます。たとえば、ユーザーのアクティビティなどの関連イベントを 1 つのレコードに結合して、ダウンストリーム分析を行います。
キーの並列処理: ベンチマークでは、均等に分散された 1,000,000 個のキーを使用しました。
セマンティクス: パイプラインは exactly-once モードを使用してテストされました。集計では、正確性を確保し、グループとウィンドウ内での重複カウントを防ぐために、exactly-once セマンティクスが必要です。
ジョブ構成
次の表に、Dataflow ジョブの構成方法を示します。
| 設定 | マップのみ、exactly-once | マップのみ、at-least-once | ウィンドウ集計、exactly-once |
|---|---|---|---|
| ワーカーのマシンタイプ | n1-standard-2 |
n1-standard-2 |
n1-standard-2 |
| ワーカー マシンの vCPU | 2 | 2 | 2 |
| ワーカー マシンの RAM | 7.5 GiB | 7.5 GiB | 7.5 GiB |
| ワーカー マシンの Persistent Disk | 標準永続ディスク（HDD）、30 GB | 標準永続ディスク（HDD）、30 GB | 標準永続ディスク（HDD）、30 GB |
| 初期ワーカー数 | 70 | 30 | 180 |
| 最大ワーカー数 | 100 | 100 | 250 |
| Streaming Engine | はい | ○ | はい |
| 水平自動スケーリング | はい | ○ | はい |
| 課金モデル | リソースベースの課金 | リソースベースの課金 | リソースベースの課金 |
| Storage Write API（gRPC）は有効になっていますか？ | はい | ○ | はい |
| Storage Write API（gRPC）ストリーム | 200 | 該当なし | 500 |
| Storage Write API（gRPC）のトリガー頻度 | 5 秒 | 該当なし | 5 秒 |
ストリーミング パイプラインには、BigQuery Storage Write API（gRPC）を使用することをおすすめします。Storage Write API（gRPC）で exactly-once モードを使用する場合は、次の設定を調整できます。
書き込みストリームの数 。書き込みステージで 十分なキーの並列処理 を確保するには、Storage Write API（gRPC）ストリームの数をワーカー CPU の数より大きい値に設定し、BigQuery 書き込みストリームのスループットを適切なレベルに維持します。
トリガー頻度 。スループットの高いパイプラインには、1 桁の秒単位の値が適しています。
詳細については、 Dataflow から BigQuery に書き込むをご覧ください。
ベンチマークの結果
このセクションでは、ベンチマーク テストの結果について説明します。
スループットとリソース使用量
次の表に、パイプラインのスループットとリソース使用量のテスト結果を示します。
| 結果 | マップのみ、exactly-once | マップのみ、at-least-once | ウィンドウ集計、exactly-once |
|---|---|---|---|
| ワーカーあたりの入力スループット | 平均: 17 MBps、n=3 | 平均: 21 MBps、n=3 | 平均: 6 MBps、n=3 |
| すべてのワーカーの平均 CPU 使用率 | 平均: 65%、n=3 | 平均: 69%、n=3 | 平均: 80%、n=3 |
| ワーカーノードの数 | 平均: 57、n=3 | 平均: 48、n=3 | 平均: 169、n=3 |
| Streaming Engine コンピューティング単位数（1 時間あたり） | 平均: 125、n=3 | 平均: 46、n=3 | 平均: 354、n=3 |
自動スケーリング アルゴリズムは、ターゲット CPU 使用率レベルに影響する可能性があります。ターゲット CPU 使用率を高くまたは低くするには、自動スケーリング範囲またはワーカー使用率のヒントを設定します。使用率のターゲットが高いと、費用は削減されますが、特に負荷が変動する場合、テールレイテンシが悪化する可能性があります。
ウィンドウ集計パイプラインの場合、集計のタイプ、ウィンドウ サイズ、キーの並列処理は、リソース使用量に大きな影響を与える可能性があります。
レイテンシ
次の表に、パイプライン レイテンシのベンチマーク結果を示します。
| ステージのエンドツーエンドの合計レイテンシ | マップのみ、exactly-once | マップのみ、at-least-once | ウィンドウ集計、exactly-once |
|---|---|---|---|
| P50 | 平均: 800 ミリ秒、n=3 | 平均: 160 ミリ秒、n=3 | 平均: 3,400 ミリ秒、n=3 |
| P95 | 平均: 2,000 ミリ秒、n=3 | 平均: 250 ミリ秒、n=3 | 平均: 13,000 ミリ秒、n=3 |
| P99 | 平均: 2,800 ミリ秒、n=3 | 平均: 410 ミリ秒、n=3 | 平均: 25,000 ミリ秒、n=3 |
テストでは、3 回の長時間実行テストで、ステージごとのエンドツーエンドのレイテンシ
（job/streaming_engine/stage_end_to_end_latencies
指標）を測定しました。この指標は、Streaming Engine が各パイプライン ステージで費やす時間を測定します。これには、次のようなパイプラインの内部ステップがすべて含まれます。
- 処理のためのメッセージのシャッフルとキューイング
- 実際の処理時間（メッセージを行オブジェクトに変換するなど）
- 永続状態の書き込みと、永続状態の書き込みのキューイングに費やされた時間
もう 1 つのレイテンシ指標は データの更新速度です。ただし、データの更新速度は、ユーザー定義のウィンドウ処理やソースの上流の遅延などの要因によって影響を受けます。システム レイテンシは、負荷時のパイプラインの内部処理効率と健全性の客観的なベースラインを提供します。
データは実行ごとに約 1 日間測定され、安定した定常状態のパフォーマンスを反映するために、最初の起動期間は除外されました。結果には、追加のレイテンシが発生する 2 つの要因が示されています。
exactly-once モード。exactly-once セマンティクスを実現するには、重複除去のために決定論的なシャッフルと永続状態のルックアップが必要です。at-least-once モードでは、これらのステップをバイパスするため、大幅に高速に実行されます。
ウィンドウ集計。ウィンドウを閉じる前に、メッセージを完全にシャッフルしてバッファリングし、永続状態に書き込む必要があります。これにより、エンドツーエンドのレイテンシが増加します。
ここに示されているベンチマークはベースラインを表しています。レイテンシはパイプラインの複雑さに大きく影響されます。カスタム UDF、追加の変換、複雑なウィンドウ処理ロジックはすべてレイテンシを増加させる可能性があります。合計やカウントなどの単純で削減率の高い集計は、リストへの要素の収集などの状態を保持するオペレーションよりもレイテンシが低くなる傾向があります。
費用を見積もる
料金計算ツールを使用して、 リソースベースの課金で、同等のパイプラインのベースライン費用を見積もることができます。 Google Cloud 手順は次のとおりです。
- 料金計算ツールを開きます。 料金計算ツール。
- [Add To Estimate] をクリックします。
- Dataflow を選択します。
- [サービスタイプ] で [Dataflow Classic] を選択します。
- [高度な構成] を選択して、オプションの完全なセットを表示します。
- ジョブを実行する場所を選択します。
- [サービスの種類] で [Streaming] を選択します。
- [Enable Streaming Engine] を選択します。
- ジョブの実行時間、ワーカーノード、ワーカーマシン、 Persistent Disk ストレージの情報を入力します。
- Streaming Engine コンピューティング単位数の推定数を入力します。
リソースの使用量と費用のスケールは、入力スループットにほぼ比例しますが、ワーカー数が少ない小規模なジョブの場合、総費用は固定費が大部分を占めます。開始点として、ベンチマーク結果からワーカーノードの数とリソース消費量を推定できます。
たとえば、入力データレートが 100 MiB/秒の exactly-once モードでマップのみのパイプラインを実行するとします。1 GiB/秒のパイプラインのベンチマーク結果に基づいて、リソース要件を次のように見積もることができます。
- スケーリング ファクタ:（100 MiB/秒）/（1 GiB/秒）= 0.1
- 予測されるワーカーノード: 57 ワーカー × 0.1 = 5.7 ワーカー
- 1 時間あたりの Streaming Engine コンピューティング単位数の予測数: 125 × 0.1 = 12.5 ユニット / 時間
この値は、初期見積もりとしてのみ使用してください。実際のスループットと費用は、マシンタイプ、メッセージ サイズの分布、ユーザーコード、集計タイプ、キーの並列処理、ウィンドウ サイズなどの要因によって大きく異なる場合があります。詳細については、 Dataflow の費用の最適化のベスト プラクティスをご覧ください。
テスト パイプラインを実行する
このセクションでは、マップのみのパイプラインの実行に使用された
gcloud dataflow flex-template run
コマンドを示します。
exactly-once モード
gcloud dataflow flex-template run JOB_ID \
--template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/PubSub_to_BigQuery_Flex \
--enable-streaming-engine \
--num-workers 70 \
--max-workers 100 \
--parameters \
inputSubscription=projects/PROJECT_IDsubscriptions/SUBSCRIPTION_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
useStorageWriteApi=true,\
numStorageWriteApiStreams=200 \
storageWriteApiTriggeringFrequencySec=5
at-least-once モード
gcloud dataflow flex-template run JOB_ID \
--template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/PubSub_to_BigQuery_Flex \
--enable-streaming-engine \
--num-workers 30 \
--max-workers 100 \
--parameters \
inputSubscription=projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME,\
outputTableSpec=PROJECT_ID:DATASET.TABLE_NAME,\
useStorageWriteApi=true \
--additional-experiments streaming_mode_at_least_once
次のように置き換えます。
JOB_ID: Dataflow ジョブ IDPROJECT_ID: プロジェクト IDSUBSCRIPTION_NAME: Pub/Sub サブスクリプションの名前DATASET: BigQuery データセットの名前TABLE_NAME: BigQuery テーブルの名前
テストデータを生成する
テストデータを生成するには、次のコマンドを使用して ストリーミング データ ジェネレータ テンプレートを実行します。
gcloud dataflow flex-template run JOB_ID \
--template-file-gcs-location gs://dataflow-templates-us-central1/latest/flex/Streaming_Data_Generator \
--num-workers 70 \
--max-workers 100 \
--parameters \
topic=projects/PROJECT_ID/topics/TOPIC_NAME,\
qps=1000000,\
maxNumWorkers=100,\
schemaLocation=SCHEMA_LOCATION
次のように置き換えます。
JOB_ID: Dataflow ジョブ IDPROJECT_ID: プロジェクト IDTOPIC_NAME: Pub/Sub トピックの名前SCHEMA_LOCATION: Cloud Storage 内のスキーマ ファイルのパス
ストリーミング データ ジェネレータ テンプレートは、JSON データ ジェネレータ ファイルを使用してメッセージ スキーマを定義します。ベンチマーク テストでは、次のようなメッセージ スキーマを使用しました。
{ "logStreamId": "{{integer(1000001,2000000)}}", "message": "{{alphaNumeric(962)}}" }
次のステップ
- Dataflow ジョブ モニタリング インターフェースを使用する
- Dataflow の費用の最適化のベスト プラクティス
- ストリーミング ジョブの処理速度が遅い場合や停止している場合のトラブルシューティング
- Pub/Sub から Dataflow に読み取る
- Dataflow から BigQuery に書き込む