ストリーミングには「沈黙」という問題がある。

要約

Kafkaのパーティションが沈黙したとき、それは完了したのか、それとも単に遅いだけなのか。外から見れば両者は見分けがつかない――そしてその答えが、手元のデータに基づいて安全に行動できるかどうかを決める。Netflix、Uber、Walmartはそれぞれ対処法を公開しているが、いずれも何かを手放している。第2のパイプライン、並列処理の1レーン、リプレイ時の正確性。どれも堅実なエンジニアリングだ。しかし、Kafkaがすでに知っているただ一つのことを、誰もKafkaに尋ねていない。

私はこの問題に、フォーチュン50企業の在庫ストリームで直面した。13のKafkaトピック、各3パーティション、1日数億件のイベント、そして特定の商品に関するすべてのイベントには厳密なイベントタイム順が要求される――7日分の履歴をゼロから読み込み、そのままピーク時のライブトラフィックに突入する構成だ。最初の設計では、すべての前段に30分のウィンドウを置き、それでもウィンドウが取りこぼしたものを掃き集める第2のジョブが必要で、なお5.5%のレコードが誤った順序で出力されていた。

本番投入されたバージョンは、時計を待たない。Kafkaのオフセットを読む。

その方法を説明しよう――そしてなぜ、このアイデアがKafkaよりも古いのかも。

誰もが行っているトレードオフ / §01

これは特定のチームだけの問題ではない。4つの優れたエンジニアリング組織が、4つの堅実なトレードオフを行っている。

• Walmartは、商品と店舗の各ペアを単一のパーティションに固定し、更新するコンシューマーを常に1つに限定している。キーの固定による順序保証――並列性を犠牲にしている。
• Uberは、遅延イベントや順序違反イベントの後に再計算するため、別個のバックフィル・パイプラインを運用している。ストリーミング経路は「ウォーターマーク後に到着したイベントを破棄することで不正確さを生じうる」からだ。第2のパイプラインを犠牲にしている。
• Netflixは、Flinkパイプラインのバックフィルにおいて、リプレイされた履歴が異なる順序と暴走する状態（ステート）を生むことを発見した。正確性とステートを犠牲にしている。
• Confluent自身による、アイドル状態のKafkaパーティションへのガイダンスはこうだ。待って停止するか、無視して「パーティションが再びアクティブになった際の大量の遅延メッセージ」を受け入れるか。停止か遅延データを犠牲にしている。

どれも合理的だ。共通するのは、いずれも完了性を時間から推測していることだ――ウィンドウ、タイムアウト、待機。しかしパーティションのオフセットは、データがどこで終わるのかを正確に示している。本稿が埋めるのは、まさにそのギャップだ。

なぜストリームは入り乱れるのか / §02

Kafkaのトピックは複数のパーティションに分割され、各パーティションは独立した線路だ。パーティション内では順序が保証される。レコード1、次にレコード11、常にこの順だ。しかしパーティション間では何も調整されない――それらを読むコンシューマーが同じ位置に同時にいることは決してない。Kafkaは約束どおりに動き、コンシューマーも約束どおりに動く。それでも、順序が重要な処理にとって、結合後の出力は入り乱れている。

同じ商品について、「入荷」より先に「販売済み」に基づいて行動することはできない。したがってパイプラインが何かを解放する前に、ある問いに答えなければならない。「この時点までのすべてを受け取ったか？」。今日使われている答えのほぼすべてはタイマーだ――ウィンドウ、アイドルタイムアウト、突合ジョブ。どれも沈黙に対する推測にすぎない。

古くからある解決策：待つのではなく、報告する / §03

先例がある。そしてそれは「より良い時計」ではなかった。初期の鉄道は時隔法で運行されていた。一定の間隔が経過すれば線路は「空いているはず」なので、後続列車は出発する。

1889年6月、アイルランドのアーマー郊外の急勾配で列車が立ち往生した。切り離された後部車両が坂を逆走し、後続の列車に衝突した。約80人が死亡した。

後続列車は、規定の20分が経過した時点でアーマーを出発していた。20分が経過したのだから、線路は空いているはずだ。それが空いていないことを、アーマーにいた誰も知るすべがなかった。

英国は、より正確な懐中時計で応えたのではない。議会は商務省に、旅客鉄道における絶対閉塞方式の義務化権限を与えた。前方の信号扱所が区間の空きを確認するまで、いかなる列車もその閉塞区間に進入しない。「線路の空き」は、誰かが待ち終える時間の長さではなく、誰かが報告する事実になった。立ち往生した列車と出発済みの列車は、もはや同じには見えない。誰も沈黙から何かを読み取ろうとしなくなったからだ。

これがすべてだ。Kafkaのオフセットを読むことが、信号扱所なのだ。

実装 — 3つの部品 / §04

原則は短い。「時間は、証拠が許すときにのみ進む」。これを実現するには3つの部品が必要だった。そして難所は、人が想像する場所にはなかった。

1つ目：タイマーキューがソートそのものになる。すべてのレコードを、そのイベントタイムをキーとしてキー付きステートに待機させ、まさにそのミリ秒に対するイベントタイムタイマーを登録する。この時点では何も出力しない。ウォーターマークがそのタイムスタンプに達するとタイマーが発火し、そこに待機していたものがすべて一緒に出ていく。Flinkはキーのタイマーをタイムスタンプの昇順で発火させるため、私たちは何もソートしない。1つのメソッドと1つのステートだけだ――難しさは、ここにはなかった。

2つ目：イベントタイムを持たないストリームが、クロックになる。13のトピックのうち1つには、イベントタイムがまったくなかった――他のすべてをエンリッチするための有効な商品を配信するだけのものだ。つまりソートすべきものがなく、他のトピックにはできないことができた。処理時間からウォーターマークを発行する、つまり常に可能な限り最新のタイムスタンプを出せるのだ。ただし落とし穴がある。パイプラインのクロックは入力の最小値なので、「現在時刻」で走るストリームがあっても、履歴ストリームがまだカウント中のあいだは何も変わらない。私はこの部品を最初に作ったが、効果はゼロだった――間違っていたのではなく、順番が間違っていた。制御ストリームが意味を持つのは、他のストリームが最小値から外れたときだけであり、ストリームが最小値から外れるのはアイドルとマークされたときだ。

3つ目：ストリームの完了を決めるのはKafkaであって、タイムアウトではない。Flink組み込みのアイドル判定はタイマーだ。一定時間データがなければアイドルとマークする。しかしタイムアウトは「起動中」と「完了済み」を区別できず、起動時こそ最悪のケースだ――接続が形成される速度はまちまちで、あるトピックが静かなまま座っているとタイムアウトが発火し、他のストリームがクロックを、まだ読んでいないイベントタイムの先へと進めてしまう。すると最初のレコードは到着時点ですでに遅延している。これは遅延ではなく、違反だ。だからタイムアウトは排除しなければならなかった。起動時、レコードを1件も読む前に、ジョブはKafkaに全パーティションの最終オフセットを問い合わせ、保持する。トピックがアイドルになるのは、すべてのパーティションがその数値に到達したときだけ――大半ではなく、すべてだ。調整すべき時間の長さは存在しない。ゲートはKafkaから得た事実だからだ。

// once, at job startup
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);

// as records arrive
if (offset + 1 >= startupEndOffsets.get(partition)) markCaughtUp(partition);

得られたもの / §05

ウィンドウは閉じるまですべてのレコードを抱え込んでいた――履歴の読み込みだけで30分、しかも再起動のたびにだ。今では、ウォーターマークが通過した瞬間にレコードが解放される。平均で100ミリ秒未満、順序を崩さないために留め置かれたレコードを含めてもだ。1分目に到着したレコードは、1分目に出ていく。順序保証も同様に改善した。2層のタイムアウト式アイドル判定では5.5%のレコードが順序違反、1層なら1.5%、Kafkaの起動時オフセットで判定すればゼロ。ウィンドウは消え、掃き集めジョブも消えた――1つのジョブが7日分の履歴を読み、そのままライブ稼働する。

約束しないこと / §06

決定論的であっても、未来は予知できない。イベントがパイプラインにとって存在するのは、Kafkaに着地してからだ。プロデューサーがレコードを抱え込み、クロックがそのイベントタイムをすでに通過した後に着地した場合、そのレコードは即座に、順序を乱して出力され、違反カウンターに記録される。まだ見えていないイベントを順序付けられるストリームプロセッサーは存在しない。この境界の内側では保証は厳密であり、境界の広さを決めるのはウォーターマークのラグだ。

同じ問いが、あらゆる場所に / §07

一度気づけば、同時に到着しないストリームから真実を判断するあらゆるシステムで、この問いが見えてくる。

河川の水位計が洪水の最中に沈黙する。川は安定しているのか、それとも水がセンサーに達したのか。2018年、USGSはハリケーン・フローレンスで観測所がオフラインになったと報告した――最も重要な観測所がだ。その沈黙を「便りがないのは良い便り」と読むのは誤りだ。

カメラリグ。複数のカメラでアスリートを撮影すると、各カメラは独立したストリームになる。まるでパーティションのように。すべてのカメラがその瞬間を報告するまで、3D姿勢を融合することはできない。沈黙したカメラは、フレームを落としたのかもしれないし、静止している人物を映しているだけかもしれない。i3d.aiでは、同じ問いが毎秒数十回繰り返されている。

エージェント。AIエージェントがツールを呼び出し、何も返ってこない。エージェントは沈黙を成功として扱うのが非常に得意だ――エラーが返ってこなかったのだから、機械の速度で次のステップへ。1時間前の状態のまま顧客アカウントに対して行動するエージェントは、タイマーで発車させられた列車と同じだ。解決策は鉄道が見つけたものと同じである。いかなるステップも沈黙を根拠に進めない。すべてのステップは、指し示せる確認を根拠に進む。

ストリーミングにおける最も安上がりな信頼性向上は、より高速なエンジンでも、より大きなウィンドウでもない。沈黙から「完了」を推測するのをやめ、すでに知っているシステムに尋ねることだ。ウィンドウはすべてのレコードについて最悪ケースを待ち切る。オフセット確認は、証拠が要求する時間だけ待ち、1ミリ秒たりとも余分には待たない。

簡素化したリファレンス実装――ジェネレーター、ハンドラー、キャッチアップトラッカー――を次回公開する予定だ。