この記事は TVer Advent Calendar 2025 19 日目の記事です。昨日、18 日目の記事は@k0bya4 さんの、「30分で Spanner の検索とグラフクエリを試す」でした。
はじめに
TVer 広告プロダクト本部 SRE の髙品（@datahaikuninja）です。広告プロダクトを支えるシステム全体の信頼性向上と、広告配信システムのバックエンド開発を担当しています。
私たちは、ユーザーのセグメント（属性）データベースとして Google Cloud の Bigtable を採用しています。広告配信時に Bigtable にアクセスして広告リクエストを送信したユーザーのセグメントを取得し、セグメントターゲティング配信を実現しています。
Bigtable にセグメントデータを書き込む ETL で、BigQuery のリバース ETL という機能を使っているのですが、リバース ETL ジョブの同時実行数が増加して以下の事象が発生することがありました。
- BigQuery の計算資源であるスロットの取り合い（スロット競合）
- Bigtable への書き込みスロットリング
(1) はセグメントデータ書き込み処理にかかる時間増加につながります。BigQuery は実行中のクエリジョブが均等にスロットを使用できるように調整するフェアスケジューリング1を行いますが、同時に多数のジョブが実行されるとスロットを要求するジョブが増えて、結果として各ジョブの実行完了に時間がかかるようになります。
(2) はリバース ETL ジョブ自体の失敗、という問題を引き起こします。多数のリバース ETL ジョブが大量の書き込みリクエストを Bigtable に送信すると、Bigtable サービスの書き込みキューが溢れてスロットリングを起こし2、BigQuery のリバース ETL ジョブはエラーで終了します。
これらの問題を解決するために ETL をリアーキテクチャして、リバース ETL ジョブの同時実行数を制御できるようにしました。この記事では、ETL のリアーキテクチャと、同時実行数制御のコアである Google Cloud の Pub/Sub を使った Worker Pool パターンの実装を紹介します。
Bigtable にセグメントデータを書き込む ETL は、Cloud Storage のファイル作成イベントにより開始します。リアーキテクチャ前は以下の構成でした。
graph LR
A[Cloud Storage] -->|ファイル作成イベント| B[Eventarc]
B -->|ワークフロー起動| C[Workflows]
C -->|ジョブ作成| D[Cloud Run Jobs]
D -->|クエリジョブ実行| E[BigQuery]
E -->|リバースETL実行| F[Bigtable]
- Cloud Storage にセグメントデータ (CSV ファイル) がアップロードされると、Eventarc にファイル作成イベント（google.cloud.storage.object.v1.finalized）が送信されます。
- Eventarc が Cloud Workflows を起動します。
- Workflows が Cloud Run Jobs にジョブを作成します。Workflows ではファイルパスを検証してジョブを作成するかどうか判断し、ファイルパスに応じて適切な環境変数をセットして Cloud Run Jobs の構成をオーバーライドしつつ呼び出します。
- Cloud Run Jobs で実行されるアプリケーションが BigQuery にクエリジョブ（リバース ETL ジョブ）を作成します。
- BigQuery が 1 でアップロードされた CSV ファイルを外部テーブルとして参照してデータを検証、変換して、Bigtable へリバース ETL を実行します。
BigQuery のリバース ETL を使うと Bigtable への ETL のコア処理をユーザーが自前実装する必要がなく便利です。Cloud Run Jobs で実行されるアプリケーションで同じことをやるとしたら、単にクエリジョブを発行するだけでなく以下の機能を自分で実装しなければいけません。
BigQuery のリバース ETL を使えば上記の処理は GoogleSQL を書くだけで実装できます。処理のほとんどを BigQuery にオフロードしており、Cloud Run インスタンスの CPU とメモリはほとんど使用しないのでリソース管理から解放される点もメリットです。
基本的にはよく動く設計なのですが、Cloud Storage に同時に多数のセグメントデータがアップロードされるとリバース ETL ジョブも同時に多数実行されてしまうので、冒頭で紹介した BigQuery のスロット競合と Bigtable の書き込みスロットリングが発生することがあります。注意深く書いておきますが、これらの問題は常に発生するわけではありません。例えば、CSV ファイルのレコード数が少ない場合は問題が発生する可能性は低いです。1つのリバース ETL ジョブが更新する Bigtable の行数が少ないので、ジョブがすぐに終了して同時実行数が増えないからです。実際、運用初期は問題は発生していませんでした。しかし、現在の私たちのシステムでは数百万から数千万レコードの CSV ファイルが Cloud Storage に多数同時にアップロードされます。経験上、このような場合に BigQuery のスロット競合と Bigtable の書き込みスロットリングに遭遇します。Bigtable のテーブルを数百万行以上更新する重いジョブの処理には比較的時間がかかるので同時実行数が増えてしまっています。
問題を解決するために、以下の構成にリアーキテクチャしました。
graph LR
%% Nodes
A["Cloud Storage"]
B["Eventarc"]
C["Workflows"]
D["Pub/Sub"]
E["GKE Deployment<br>(Worker)"]
F["BigQuery"]
G["Bigtable"]
%% Edges
A -->|"ファイル作成イベント"| B
B -->|"ワークフロー起動"| C
C -->|"メッセージ（ジョブ）をプッシュ"| D
E -->|"メッセージ（ジョブ）をプル"| D
E -->|"クエリジョブ実行"| F
F -->|"リバースETL<br>実行"| G
Workflows は Cloud Run Jobs を呼び出してジョブを作成するのではなく、Pub/Sub にメッセージ（ジョブ）を push します。そして、GKE Deployment として常時起動しているワーカーが Pub/Sub に push されたメッセージを pull し、BigQuery にリバース ETL ジョブを作成します。ワーカーは同時に処理するメッセージ数を制限することで、BigQuery に作成されるリバース ETL ジョブの同時実行数を制御することが可能です。リアーキテクチャにより、BigQuery と Bigtable のリソースを使いすぎないようにジョブを処理できるようになったので、問題が発生しなくなりました。
ちなみに、Bigtable へのリバース ETL を実行するには QUERY の割当があるスロット予約が必要なので、スロット予約を使用するジョブの同時実行数制限によってリバース ETL ジョブの同時実行数がコントロールできないか試したのですが、うまくいきませんでした3。できないのなら、アプリケーションで制御すればよいと考えていたため、サポートへの問い合わせはしていません。
Pub/Sub は「メッセージを生成するサービスを、それらのメッセージを処理するサービスと切り離す、非同期のスケーラブルなメッセージングサービス」と紹介されています4。リアーキテクチャ前から、Google Cloud のリソースレベルではジョブの作成 (Workflows) とジョブの処理 (Cloud Run Jobs) は切り離されていましたが、ジョブの生成と処理はタイミングとしては切り離されておらず、ジョブが生成されると直ちに処理されていました。タイミングが分離されていないことが、ジョブの同時実行数が増えてしまう主要因だったと言ってもよいと思います。リアーキテクチャ後は Pub/Sub が Workflows と GKE の間に挟まり、ジョブの生成と処理のタイミングを切り離すジョブキューの役割を担います。
Pub/Sub をジョブキューとして利用するときに注意したことがあります。ワーカーが Pub/Sub に push されたメッセージ（ジョブ）を直ちに pull して処理してしまうと、BigQuery のリバース ETL ジョブの同時実行数を制限することができません。この記事で扱っている問題を解決するワーカーを開発するときは、Pub/Sub からジョブを pull するタイミングを制御すること、つまり同時に処理するジョブ数を制限することが重要です。
Pub/Sub を使った Worker Pool パターン実装
同時に処理するジョブ数を制限するというテーマは Go の並行処理パターンの議論で扱われます。Go の channel と goroutine を組み合わせて goroutine (ワーカー) 数を制御するやり方は、Worker Pool という名前で知られるパターンです。以下では、 Pub/Sub を利用した Worker Pool 実装を紹介します。Go のサンプルコードと並行処理パターンですが他の言語にも応用できる内容だと思います。
まず、Pub/Sub を使った Worker Pool 開発において知っておく必要がある Pub/Sub サービスの用語5を導入します。
- パブリッシャー (Publisher)
- メッセージ送信者
- この記事では Workflows が Publisher です。ジョブを Pub/Sub メッセージで表現します。
- トピック (Topic)
- サブスクリプション (Subscription)
- トピックに対して1つ以上存在するメッセージを受信するための登録
- サブスクライバー (Subscriber)
- Pub/Sub クライアントライブラリ
つまり、Pub/Sub はトピックを通じてパブリッシャーとサブスクライバーがメッセージをやりとりするサービスです。開発者は公式のクライアントライブラリを使用してサブスクライバーアプリケーションを開発することができます。
次に、サブスクリプションの種類6を導入します。
この記事で紹介する Worker Pool で使用するのは pull サブスクリプションです。以下は、pull サブスクリプションにおけるメッセージの送受信のモデルです。
sequenceDiagram
participant P as Pub/Sub<br>Pull Subscription
participant S as Subscriber
%% 右から左へのリクエスト
S->>P: StreamingPullRequest
S->>P: StreamingPullRequest
%% 左から右へのレスポンス
P->>S: StreamingPullResponse
P->>S: StreamingPullResponse
%% (N)Ack
S->>P: (N)AckRequest
P->>S: Empty
Pub/Sub サーバーとサブスクライバー間の通信が、(1) PullRequest (2) PullResponse (3) (N)AckRequest (4) Empty ではないことが分かりづらいかもしれません。Pub/Sub クライアントライブラリにはメッセージを1つずつ受信して確認する、同期 pull モードのための単項 Pull (unary pull) API も存在しますが、 代わりに、以下の理由で図中に示した Streaming Pull API の使用が推奨されています7。
- 単項 Pull API を使うためには低レベルのクライアントライブラリを使用することになり、高度なメッセージ配信制御機能を自分でコーディングしなければならない
- Streaming Pull API の方がメッセージの配信レイテンシーを最小化し、スループットを最大化できる
今回のリアーキテクチャで実装したワーカーアプリケーションにおいては、低レイテンシーと高スループットは必須ではありませんが、「1 回限りの配信」や「フロー制御」、「リース管理」のような高度な配信制御機能は必要でした。そのため、 Streaming Pull API に対応した高レベルクライアントライブラリを使用していますが、メッセージの受信設定を調整すれば単項 Pull API を使わなくても同時処理メッセージ数制限は可能なので問題ありません。
ようやくですが、サンプルコードの紹介です。Pub/Sub エミュレーターを利用して、同時処理メッセージ数を制限する Pub/Sub サブスクライバーの動作をローカルで試せるサンプルプロジェクトを作成しました。以下に置いてあります。
https://github.com/datahaikuninja/google-cloud-pubsub-worker-pool-sample
Pub/Sub サブスクライバーが同時に処理するメッセージ数を制限するには、Pub/Sub クライアントライブラリを使用するサブスクライバーアプリケーションのメッセージの受信設定を調整します。Go では pubsub.Subscriber 型の構造体が公開する ReceiveSettings.MaxOutstandingMessages フィールドの値になります。以下が構造体の定義です。
https://github.com/googleapis/google-cloud-go/blob/ce28733097806539575dafcd019920f340c83304/pubsub/v2/subscriber.go#L124-L129
サンプルコードでは以下の箇所です。
func NewWorker(ctx context.Context, pubsub *pubsub.Client, subscription string) *Worker {
subscriber := pubsub.Subscriber(subscription)
subscriber.ReceiveSettings.MaxOutstandingMessages = maxOutstanding
return &Worker{
subscriber: subscriber,
}
}
MaxOutstandingMessages は最大未処理メッセージ数という意味ですから、Pub/Sub サブスクライバーが Ack （確認・処理）していないメッセージ数が MaxOutstandingMessages に達すると Pub/Sub サブスクライバーは Pub/Sub サーバーからのメッセージ取得を一時停止します。
Pub/Sub サブスクライバーは Subscriber.Receive() で Pub/Sub サーバーからメッセージを取得し、このメソッドの第 2 引数で渡されたコールバック関数を実行します。以下がクライアントライブラリ側の実装です。
https://github.com/googleapis/google-cloud-go/blob/main/pubsub/v2/subscriber.go#L197
Subscriber.Receive() はメッセージを受信するたびにコールバック関数を新しい goroutine で実行するので、受信して処理中のメッセージ数 = コールバック関数を実行する goroutine 数の関係です8。MaxOutstandingMessages に達してメッセージ受信を一時停止すると、コールバック関数を実行する新しい goroutine は起動しません。これが、Pub/Sub クライアントの同時処理メッセージ数制限の仕組みであり、Pub/Sub を使った Worker Pool のコアです。
サンプルコードでは Worker.Run() で Subscriber.Receive() を包んでワーカーを表現しています。Subscriber.Receive() の第 2 引数に渡したコールバック関数を実行する goroutine こそワーカーの実体と言えるものであり、コールバック関数がワーカーに処理させたいビジネスロジックです。
func (w *Worker) Run(ctx context.Context) error {
var err error
var wg sync.WaitGroup
var workerCnt atomic.Uint64
err = w.subscriber.Receive(ctx, func(ctx context.Context, m *pubsub.Message) {
wg.Add(1)
workerCnt.Add(1)
log.Printf("W: subscriber.Receive spawn worker. worker count: %v", workerCnt.Load())
log.Printf("W: worker pulled messageID: %s", m.ID)
defer func() {
workerCnt.Add(^uint64(0))
log.Printf("W: worker exiting. worker count: %v", workerCnt.Load())
wg.Done()
}()
err := w.processJob(m)
if err != nil {
log.Printf("W: worker failed to process messageID: %s, jobID: %s, error: %v", m.ID, m.Attributes["jobID"], err)
m.Nack()
return
}
m.Ack()
log.Printf("W: worker acked messageID: %s, jobID: %s", m.ID, m.Attributes["jobID"])
})
if err != nil && !errors.Is(err, context.Canceled) {
return fmt.Errorf("W: unexpected error in subscriber.Receive: %v", err)
} else {
log.Println("W: subscriber.Receive exited by context.Canceled")
}
wg.Wait()
log.Printf("W: worker count: %v", workerCnt.Load())
log.Println("W: all worker finished, exiting")
return nil
}
サンプルプロジェクトを動かしてみると、W: subscriber.Receive spawn worker. worker count: %v のログが、コールバック関数を実行する goroutine の数（ワーカーの数）が MaxOutstandingMessages を超えないことを示しています。
2025/12/15 10:31:15 W: starting subscriber.Receive
2025/12/15 10:31:16 P: pushed message
2025/12/15 10:31:16 W: subscriber.Receive spawn worker. worker count: 1
2025/12/15 10:31:16 W: worker pulled messageID: 1
2025/12/15 10:31:16 W: processing messageID: 1, jobID: 31bd5a71-7c17-46a4-bf81-8d6dc0983fae
2025/12/15 10:31:16 W: read message data: heavy job!
2025/12/15 10:31:17 P: pushed message
2025/12/15 10:31:17 W: subscriber.Receive spawn worker. worker count: 2
2025/12/15 10:31:17 W: worker pulled messageID: 2
2025/12/15 10:31:17 W: processing messageID: 2, jobID: 180ceff8-5fcd-455b-a4c5-e787380d6b95
2025/12/15 10:31:17 W: read message data: heavy job!
2025/12/15 10:31:18 P: pushed message
2025/12/15 10:31:18 W: subscriber.Receive spawn worker. worker count: 3
2025/12/15 10:31:18 W: worker pulled messageID: 3
2025/12/15 10:31:18 W: processing messageID: 3, jobID: 9739e63b-d0e2-4eca-8407-427c0507cbe3
2025/12/15 10:31:18 W: read message data: heavy job!
2025/12/15 10:31:19 P: pushed message
2025/12/15 10:31:20 P: pushed message
2025/12/15 10:31:21 P: pushed message
2025/12/15 10:31:21 W: worker acked messageID: 1, jobID: 31bd5a71-7c17-46a4-bf81-8d6dc0983fae
2025/12/15 10:31:21 W: worker exiting. worker count: 2
2025/12/15 10:31:21 W: subscriber.Receive spawn worker. worker count: 3
2025/12/15 10:31:21 W: worker pulled messageID: 4
2025/12/15 10:31:21 W: processing messageID: 4, jobID: d9441820-cd5b-45ac-b103-a513d41269b0
2025/12/15 10:31:21 W: read message data: heavy job!
2025/12/15 10:31:22 P: pushed message
2025/12/15 10:31:22 W: worker acked messageID: 2, jobID: 180ceff8-5fcd-455b-a4c5-e787380d6b95
2025/12/15 10:31:22 W: worker exiting. worker count: 2
2025/12/15 10:31:22 W: subscriber.Receive spawn worker. worker count: 3
2025/12/15 10:31:22 W: worker pulled messageID: 5
2025/12/15 10:31:22 W: processing messageID: 5, jobID: 5142303e-5c84-40e0-8f82-660f2695fcea
2025/12/15 10:31:22 W: read message data: heavy job!
2025/12/15 10:31:23 P: pushed message
2025/12/15 10:31:23 W: worker acked messageID: 3, jobID: 9739e63b-d0e2-4eca-8407-427c0507cbe3
2025/12/15 10:31:23 W: worker exiting. worker count: 2
以上が、Pub/Sub を使った Worker Pool の実装でした。
Pub/Sub を使わない他の Worker Pool 実装も紹介して、Pub/Sub を使う Worker Pool のメリットを考えてみます。
Go の Worker Pool パターンのコンセプトは、固定数の goroutine が channel から job や task を取得して処理する、という構造です。
Worker Pool の基本形は Go By Examples: Worker Pools で紹介されています。このサンプルコードでは、3 つの Worker が jobs channel からジョブを取り出して処理します。Worker の数を超える job が channel に投入されますが、総 job 数と jobs channel のバッファサイズが一致するように書いているのでブロッキングは起こりません。バッファサイズを超える job を送信すると送信側がブロックされます。
より高度な Worker Pool が Dispatcher-Worker です。Handling 1 Million Requests per Minute with Golang という Blog で紹介された実装が有名だと思います。
Dispatcher-Worker では、Worker は job を直接取得はせず、Dispatcher が空いている Worker に job を割り振ります。特徴的なのは、各ワーカーが専用のジョブ受信 channel を持ち、ジョブを処理し終えると自身のジョブ受信 channel をワーカープールの channel に向けて送信し、Dispatcher は ワーカープールの channel が受信した各ワーカーのジョブ受信 channel に job を送信する、という点です。ジョブ受信 channel をやりとりするワーカープール channel という発想が Dispatcher-Worker のコアだと思います。Dispatcher-Worker のサンプルコードを説明すると記事が長くなるので引用は控えます。私が、コアであると述べた特徴を意識して WorkerPool chan chan Job の初期化と利用方法に注目して参照元 Blog を読んでもらえたらと思います。翻訳記事もあります。
シンプルな Worker Pool, Dispatcher-Worker と Pub/Sub を使う Worker Pools の違いは主に2つあると思います。
- Pub/Sub トピックを job queue として利用することによるメリットがある
- キューサイズを事前に考える必要がない
- ジョブ消失リスクが低い
- Pub/Sub クライアントライブラリの Subscriber が Dispatcher 相当の機能を備えているので、開発者はジョブを処理するビジネスロジックの実装に集中できる
Go の channel で job queue をつくるときは、channel のバッファサイズと、アプリケーションシャットダウンによるジョブ消失について考慮しておくべきだと思います。まず、バッファサイズは十分な大きさにしておかないとバッファが溢れて channel への送信がブロックされます。Dispatcher-Worker のように外部からジョブを受け取る窓口 (job collector) を作るなら channel の長さを調べてバッファが埋まっていたら呼び出し元にリトライを促したりすることはできますが、妥当なバッファサイズはいくつなのか決めるのは個人的には悩ましいです。事実上溢れないバッファサイズにすることは可能ですが、無駄に多くのメモリを確保することになるので非効率です。そして、オンメモリの job queue の中身は想定外のアプリケーションシャットダウンによって消失するリスクがあります。SIGTERM シグナルをフックしてジョブをどこかに退避したり、job の新規受付を停止して job queue が空になるまでシャットダウンを遅らせるといった Graceful Shutdown のアイディアは浮かびますが、OOM による SIGKILL や基盤の障害でいきなりプロセスが死んでしまう場合には無力ですし、Graceful Shutdown するためにプロセスの終了を遅延させるにも限度があります。
Pub/Sub を job queue として利用するときは事前にバッファサイズを設計する必要はありません。その代わりに Pub/Sub の割り当て (quota)9 を超えないか事前に確認します。相当大規模なシステムでない限り Pub/Sub の push/pull スループットの割り当てを超えてしまうことはないので、設計段階で割り当てを心配しすぎる必要はないです。私たちは、Worker Pool の他に、広告配信サーバーが記録する広告ログの送信先としても Pub/Sub を利用しており、BigQuery サブスクリプションでログを非同期で BigQuery に取り込んでいます。広告配信サーバーは 1 日に億単位の広告リクエストを処理して配信ログを Pub/Sub へ push していますが、Pub/Sub の割当を超えてエラーが発生したことはありません。
ジョブ消失については、バッファ溢れから割り当て超過のように考慮点が変わるのではなく、Pub/Sub を使うことでリスクを下げることができます。Pub/Sub に push された ジョブは、基本的にはサブスクライバーが確認応答期限までに Ack 応答をサーバーに返さない場合に再配信されます。この仕組みを利用すれば、想定外のシャットダウンが起きてもアプリケーションが再起動したときに Pub/Sub から未処理ジョブの再配信を受けることができます。
Pub/Sub クライアントライブラリに Dispatcher 相当のワーカー管理を任せられることも、目立たないですが確かなメリットだと思います。サンプルプロジェクトのコードを見てもらえれば分かるように、Subscriber.Receive() に渡すコールバック関数を書くだけで堅牢な Worker Pool を実装できるので、本質的なビジネスロジックの実装に集中することができます。
なお、私はこの記事において、どんな場合でも Pub/Sub を使った Worker Pool 実装が他に優っていると言いたいのではありません。個人的には、できる限りマネージドサービスを組み合わせてアーキテクチャとアプリケーションを設計するのが好みであり、Pub/Sub を使うと比較的楽にプロダクションで使える Worker Pool を実装できた、という事例を紹介したいと思ってこの記事を書いています。goroutine と channel を使ったシンプルな実装で十分な場合もあるでしょうし、制約条件で Pub/Sub が合わない場合はマネージドサービスを諦めて自前実装する場合もあると思います。個人の価値判断は含まれていますが、記事の目的は設計パターンの紹介であり、異なるパターンの優劣を付けることではありません。
おわりに
この記事では、Bigtable にデータを書き込む ETL に Pub/Sub を使った Worker Pool を組み込むことで、BigQuery のリバース ETL ジョブの同時実行数を制限し、ETL の信頼性を向上させる方法を紹介しました。記事を書いてみて気付いたのは、Pub/Sub を使った非同期処理パターンの汎用性です。BigQuery のリバース ETL は、Bigtable だけでなく Spanner にも対応しているので、この記事で紹介したアーキテクチャは Spanner にも応用できそうですし、BigQuery のリバース ETL に限らず、データベースの更新等の重いジョブの同時実行数を制御したいという場面では役立ちそうです。低レイテンシー・高スループットが求められる大量の非同期ジョブの処理においても、もちろん有効なアーキテクチャだと思います。
記事で扱う範囲が広がってしまうので Worker Pool の実行環境については敢えて触れませんでした。一言二言だけ書いておくと、コンテナアプリケーションとして Worker Pool をデプロイするなら 2025 年 12 月時点では GKE と Cloud Run Worker Pools の2つの選択肢があります。従来、GKE を使っていないユーザーは常時起動ワーカーを実行するためのベストな選択肢がありませんでした。GKE を使っていないユーザーにとっては、常時起動ワーカーを実行するためだけに GKE を採用するのはオーバーエンジニアリングだと思いますし、Cloud Run Services では制約が多く、Cloud Run Jobs ではそもそも不適切な使用方法になってしまうという悩みがありました10。しかし、Cloud Run Worker Pools の登場により Cloud Run 中心のアーキテクチャを変更することなく、常時起動ワーカーを実行することが可能になっています。2025 年 12 月時点では、Cloud Run Worker Pools はまだプレビューですが、GA が楽しみですね。
長くなってしまいましたが、ここまで読んでいただきありがとうございます。この記事が、読者の方の良き設計の参考になると嬉しいです。明日は @iideshoさんの「TVerとテレビと私（今回は車載多め）2025」です。
