ストリーミングとリアルタイム分析

リアルタイム取り込みの選択肢と選択マップ

第 3 章のバッチロードは「まとまった量のファイルを一括で読み込む」方式でした。これに対してストリーミングは、発生したデータを数秒〜数十秒のオーダーで BigQuery に届け続ける方式です。BigQuery でリアルタイム取り込みを実現する経路は大きく 4 つあり、どれを選ぶかはレイテンシ・変換要否・運用コストの 3 軸で決まります。

Storage Write API — ストリームの種類と exactly-once

Storage Write API（gRPC）は BigQuery への高スループットな書き込みを担う現行の中核 API です。ストリームには 4 種類あります。

• デフォルトストリーム: 作成不要ですぐ使え、書き込んだデータは即座にクエリ可能。ただし at-least-once（exactly-once が不要な高スループット用途向け）
• Committed 型（アプリケーション作成）: 書き込んだ行を即座に読み取れる点はデフォルトと同じだが、クライアントがオフセットを与えることで exactly-once を実現できる
• Pending 型（アプリケーション作成）: コミットするまで行が保留され、コミット時に原子的に可視化される。ロードジョブの代替に使うバッチ向け
• Buffered 型（アプリケーション作成）: 行単位のコミットを提供する高度なユースケース向けで、Apache Beam の BigQuery I/O コネクタなどが利用する

exactly-once は Committed ストリームで、クライアントが行ごとに付与するオフセットをサーバー側の「次に期待するオフセット」と照合する仕組みで成立します。一致した場合だけ書き込みが確定してオフセットが進み、同じオフセットでの再送は重複として無視されるため、リトライしても二重挿入が起きません。

レガシーな streaming insert（tabledata.insertAll）は現在も REST エンドポイントとして存在しますが、2026-08 時点で本書執筆にあたり確認したところ、その解説ガイドページ（docs/streaming-data-into-bigquery）は write-api-rest（Storage Write API (REST)）へ 301 リダイレクトされており、実質的に Storage Write API (REST) の文脈へ統合されています。公式ドキュメントは新規プロジェクトに対して「Storage Write API (REST) より Storage Write API (gRPC) を推奨」と明言しており、理由は料金の低さと exactly-once を含む堅牢な機能です。

書き込み直後のデータは「ストリーミングバッファ」に入ります。バッファ内のデータもクエリでは即座に参照できます。gRPC 版 Storage Write API で書き込んだ行なら、直近 30 分以内のものにも UPDATE・DELETE・MERGE などの DML を実行できます（この直近データへの DML 対応は REST 版で書き込んだデータには及ばず、マルチステートメントトランザクション内でも使えません）。バッファの中身はバックグラウンド処理で継続的にカラムナストレージへ移されます。

Storage Write API を Python から直接使う場合、既定ストリームへの最小構成はおおむね次のようになります（読み物として掲載。実行には事前に定義した Protocol Buffer 行スキーマが必要なため、本書では実装までは踏み込みません）。

from google.cloud import bigquery_storage_v1
from google.cloud.bigquery_storage_v1 import types, writer

write_client = bigquery_storage_v1.BigQueryWriteClient()
parent = write_client.table_path(PROJECT_ID, "handson_ch12", "pubsub_events")
stream_name = f"{parent}/_default"  # 既定ストリーム(at-least-once)

request_template = types.AppendRowsRequest()
request_template.write_stream = stream_name
append_rows_stream = writer.AppendRowsStream(write_client, request_template)

proto_rows = types.ProtoRows()
proto_rows.serialized_rows.append(row.SerializeToString())  # 事前定義した行スキーマ
request = types.AppendRowsRequest()
request.proto_rows = types.AppendRowsRequest.ProtoData(rows=proto_rows)
append_rows_stream.send(request)

Pub/Sub 直結・Datastream・continuous queries の使い分け

Pub/Sub BigQuery サブスクリプション

Pub/Sub のトピックに BigQuery サブスクリプションを直接アタッチすると、内部で Storage Write API (gRPC) を使って自動的にテーブルへ書き込まれます。コードは不要です。宛先テーブルの列は 2 通りの決め方があります。

• トピック／テーブルのスキーマを使わない場合: 宛先テーブルに data 列（型は BYTES・STRING・JSON のいずれか）が必須で、メッセージ本文がそのまま書き込まれます
• --use-topic-schema または --use-table-schema を指定する場合: Avro／Protocol Buffer／JSON のフィールドを列へマッピングでき、CDC（_CHANGE_TYPE / _CHANGE_SEQUENCE_NUMBER）にも対応します

--write-metadata を有効にすると、subscription_name（STRING）・message_id（STRING）・publish_time（TIMESTAMP）・attributes（STRING/JSON）が追加列として書き込まれます。配信保証は at-least-once のみで、Dataflow のように厳密な重複排除は行われません。

Datastream — RDB からの CDC

Datastream はサーバーレスの CDC（変更データキャプチャ）サービスで、MySQL・PostgreSQL（AlloyDB を含む）・Oracle・SQL Server・MongoDB・Spanner に加え、Salesforce などのアプリケーションソースにも対応します。運用データベースの変更を最小限のレイテンシで BigQuery へレプリケーションし、ほぼリアルタイムの分析基盤を組めます。

continuous queries — BigQuery 内での常駐変換

continuous queries は SQL を常駐実行し続け、書き込まれたデータをリアルタイムに変換・出力する機能です。INSERT 文で BigQuery テーブルへ、EXPORT DATA 文で Pub/Sub・Bigtable・Spanner（リバース ETL）へ出力できます。コア機能（INSERT・EXPORT DATA・生成 AI 関数呼び出し）は GA ですが、JOIN・集計・ウィンドウ集計などのステートフル操作はプレビュー段階です。前提条件として BigQuery Enterprise または Enterprise Plus エディション（第 7 章参照）が必須で、オンデマンド課金では利用できません。専用の CONTINUOUS 割り当てタイプの予約が必要で、クエリあたり最低 10 スロット、対応する予約 1 つにつき最大 500 スロットの制限があります（増枠申請可）。ユーザーアカウントで実行するジョブは最大 2 日、サービスアカウントなら最大 150 日で停止します。

ハンズオン

bq --location=US mk --dataset PROJECT_ID:handson_ch12

bq mk --table PROJECT_ID:handson_ch12.pubsub_events \
  subscription_name:STRING,message_id:STRING,publish_time:TIMESTAMP,data:STRING,attributes:STRING

gcloud pubsub topics create ch12-events

PROJECT_NUMBER=$(gcloud projects describe PROJECT_ID --format='value(projectNumber)')

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com" \
  --role="roles/bigquery.dataEditor"

gcloud pubsub subscriptions create ch12-events-to-bq \
  --topic=ch12-events \
  --bigquery-table=PROJECT_ID.handson_ch12.pubsub_events \
  --write-metadata

gcloud pubsub topics publish ch12-events --message='{"event":"ping","seq":1}'
gcloud pubsub topics publish ch12-events --message='{"event":"ping","seq":2}'

SELECT message_id, publish_time, data
FROM `PROJECT_ID.handson_ch12.pubsub_events`
ORDER BY publish_time DESC
LIMIT 10;

gcloud pubsub subscriptions delete ch12-events-to-bq
gcloud pubsub topics delete ch12-events
bq rm -r -f -d PROJECT_ID:handson_ch12

つまずきポイント

理解度チェック

まとめと次章へ

• リアルタイム取り込みは Storage Write API 直叩き・Pub/Sub 直結・Dataflow 経由・Datastream CDC の 4 経路があり、変換要否と継続反映の要件で選び分ける
• exactly-once は Storage Write API の Committed ストリームがオフセット照合で実現する。デフォルトストリームは at-least-once
• レガシー streaming insert のドキュメントは Storage Write API (REST) に統合済み。新規実装は料金・機能で勝る gRPC 版を選ぶ
• Pub/Sub BigQuery サブスクリプションはコード不要で手軽だが at-least-once。重複排除や複雑な変換が要るなら Dataflow を挟む
• continuous queries は Enterprise／Enterprise Plus エディションと専用予約が前提。コア機能は GA だが JOIN・集計はプレビュー

次章の BigQuery ML では、ここまで整えたデータ基盤を土台に、SQL だけで機械学習モデルを構築します。

• BigQuery Storage Write API の概要
• Storage Write API (gRPC) の詳細
• Pub/Sub の BigQuery サブスクリプション
• BigQuery サブスクリプションの作成手順
• BigQuery continuous queries の概要
• Datastream の概要
• BigQuery サンドボックスの制限