The challenge of stream computing engine data correctness

ストリームコンピューティングは、リアルタイムデータウェアハウス、データレイク、リアルタイムビジネス指標、ビジネスインテリジェンスなどのための高速データ構築パイプラインを提供します。一般的に、ストリームコンピューティングがもたらす主なメリットは速度（低レイテンシ）であり、次に正しいデータ結果が得られることだと認識されています。ストリームコンピューティングの分野において、正しい計算結果を得ることは容易ではありません。精巧なエンジン設計、正確なビジネスデータ定義、上流と下流のシステム間の連携などが必要となります。本稿では、ストリームコンピューティングにおけるデータ正確性の意味と、この問題を解決するために直面する課題を紹介し、ストリームコンピューティングエンジンの鍵となる設計であるストリームデータ完全性推論へと導きます。まず、完全性の形式的定義と、各種エンジンへの一般化した適用について説明します。同時に、Flink、Kafka Streams などの主流ストリームコンピューティングエンジンにおけるデータ完全性推論の設計とトレードオフも比較します。本稿がストリームコンピューティングエンジンの利用に関するより多くの知見を提供し、正しい計算データを取得するための実践的な参考資料となることを期待します。

1. ストリームコンピューティングにおける正確性

正しいデータとは、ストリームコンピューティングエンジンが計算した結果が、現実の物理世界の対象を正しく反映していることを意味します。たとえば、ユーザーがある期間に 3 回の支払いを行い、合計 100 元であった場合、「ユーザー支払い履歴合計」という指標について、ユーザーの支払い行為が完了した時点で、指標は 1 つだけでその値が 100 であることを観測できるはずです。ストリーム要素の非有界性と非順序性のため、上記の論理的推論をストリームコンピューティングで実装することは容易ではありません。指標値が 100 未満になる、複数の指標値が存在するなど、一部の「不正確な」結果が生じる可能性があります。正確な結果を得るため、実務ではストリーミングとバッチを組み合わせた Lambda アーキテクチャが依然として広く採用されています。リアルタイムで構築された増分データとバッチ処理で得られたストックデータを組み合わせ、下流のビジネスに対して迅速かつ品質管理されたデータを提供しています。

不正確なデータは、意思決定者（人または機械）のビジネスや市場に対する判断に影響を及ぼし、ビジネス判断の遅延や誤った判断を招く可能性があります。ストリーミングとバッチのコンピューティング実践は多数存在しますが、これらのソリューションは主に従来の Lambda アーキテクチャのプラットフォーム運用・メンテナンスとコンピューティングリソースの複雑さを解決するためのものであり、ストリームコンピューティング自体に起因するデータ品質問題は、ストリームエンジンの発展によっても大きく改善されていません。以下のセクションでは、ストリームコンピューティングにおけるデータ処理が不正確になる原因と、正しい結果を得るための必要十分条件を分析します。

2. 正確性の必要十分条件

ストリームコンピューティングの本質は、非同期メッセージによる分散コンピューティングです。しかし、時計同期、ネットワーク遅延、サーバー停止（GC など）などの原因により、ストリームデータ生成の論理的順序は、ストリーム処理システムに到着する物理的順序やストリーム処理システムのオペレーターと通常一致しません。ストリームコンピューティングエンジンは非順序データを処理できなければならず、そうでなければ不正確な計算結果が生じます。

2.1 一貫性は正確性と等しくない

データコンピューティングの分野には、データ一貫性とデータ正確性という 2 つの概念があり、これらは混同されやすいものですが、概念自体の観点からは、データ一貫性 ≠ データ正確性です。2 つの概念の定義は以下の通りです。

1. 正確性：ストリームコンピューティングの結果が、現実の物理世界の対象を正しく反映していること。

2. 一貫性：全フローにわたる上流と下流のシステムのデータが、同一の情報を反映していること。

一貫性は通常「exactly once」という用語と組み合わせて使用され、ストリームコンピューティングエンジンが障害から一貫性のある状態に回復でき、最終的な計算結果に重複エントリが含まれず、データが失われないことを意味します。つまり、ストリームコンピューティングエンジンの出力は、障害なしでデータが 1 回だけ処理されたかのように振る舞います。正確性の要件は一貫性の要件よりも厳格です。エンジンが一貫性のあるデータ処理を達成できなければ、正確性も達成できません。データが正しければ必ず一貫性があります。すなわち、データ一貫性はデータ正確性の必要十分条件ではなく、必要条件です。たとえば、ストリームデータソースの 4 件のレコードがエンジンの障害や不正処理により 5 回処理された場合、結果は不正確になる可能性があります。一方で、エンジンが一貫性を達成していても、ソースデータの非順序性や遅延により正しい結果が得られないこともあります。

計算の正確性は、データ完全性とエンジン一貫性によって共同で保証されます。ストリームコンピューティングプロセスを関数マッピング output = f(input) と見なすと、上記モデルにおいて、データ完全性は非有界かつ非順序なデータセットに対する制約であり、すなわち入力が定義され、エンジンデータ一貫性はデータ処理プロセス（出力を含む）f を定義します。したがって、出力は決定されます（エンジンのエンドツーエンド一貫性については、「ストリームコンピューティングエンジンのデータ一貫性の本質」を参照してください）。

2.2 ストリームデータ完全性の必要性

ストリームコンピューティングにおけるデータ完全性とは何でしょうか。ストリームコンピューティングエンジン内のデータは非有界かつ非順序であるため、この不確実なデータセットを何らかの方法で論理的な「現在パーティション」に変換し、パーティション内の確定的なデータフラグメントを分析する必要があります。「現在パーティション」は過去の一定期間（スライドウィンドウなど）でも、過去的一定数のレコードでも構いません。別の観点から見ると、ストリームコンピューティングであれバッチコンピューティングであれ、正しい計算結果を得るためには確定的な入力データセットが必要です。バッチコンピューティングモードは、1 つのパーティションが有界データセット全体をカバーする特殊なケースに過ぎません。現在、ストリームコンピューティングエンジンが「不正確」と批判される大きな理由は、既存の技術ソリューションが非有界かつ非順序なストリームデータのパーティショニングを得意としていないことにあります。

完全性推論はデータの準備完了状態を表現する手段です。入力ストリームが非順序に到着する可能性があっても、ストリームコンピューティングエンジンは不完全な入力計算を最終出力結果として使用しません。完全性は、計算エンジンが現在の計算進捗をタイムリーに追跡し、出力結果とその入力ストリームに対応する完了度を推定できることを要求します。このデータ完全性推論は、多くのストリームコンピューティングシナリオで極めて重要です。たとえば、ストリームベースのアラームシステムでは、ストリームコンピューティングエンジンは単一かつ正確なアラーム指標を生成する必要があります。事前に一部の結果を送信することは無意味であり、分散システムであるストリームコンピューティングエンジンに「アラーム指標に必要なすべてのデータが準備完了である」ことを推論する能力が求められます。また、ストリーム CEP を通じたビジネス欠損値検出のシナリオでは、完全性推論がなければ、実際のデータ欠損とデータ到着遅延の区別をつけることができません。

完全性推論はエンジン自体の状態管理にも有用です。たとえば、Apache Spark Structured Streaming と Kafka Streams は、同様の遅延アルゴリズム（イベント時間 - 固定有効期限）を使用して、計算プロセス中の状態ストレージを回収し、メモリ消費を削減しています。

3. ストリームデータ完全性の汎用ソリューション

このパートでは、元の完全性の問題から出発し、完全性の形式的表現モデルを提案し、その上でこの汎用フレームワークの下でいくつかの一般的に使用される完全性推論スキームを一般化します。このトップダウンのアプローチにより、ストリームコンピューティング分野における完全性推論の本質をより明確に理解できます。ウォーターマークなどの多くのよく知られた概念は、あるフレームワークの具体的な実装に過ぎません。より良いアプローチは、モデルの観点からそれらを抽象化すること（Why）です。フレームワークではなくモデルから考えることで、多くのエンジンの「コンセプト」に惑わされることを避けられます。

3.1 完全性推論の形式的定義

ストリームコンピューティングエンジン上で実行されるデータ処理プログラムは、一般的に有向非循環グラフとして定義されます（Timely DataFlow を除く）。グラフ内のノードはステートフルなデータ処理オペレーターを表し、有向エッジはオペレーター間のデータチャネルを表します。データフロートポロジー内のオペレーターは 1 つ以上の物理実行ノードにマッピングされ、サーバークラスター全体に分散配置されます。ストリームコンピューティングエンジンに入力される要素はイベントと呼ばれます。各イベントには 2 つの属性があります。イベント時間と処理時間です。処理時間は、イベントが処理される際のエンジンマシンのシステム時間を指します。イベント時間はイベントが発生した時刻であり、通常イベントがエンジンに到着する前に決定されます。イベントタイムスタンプは各イベントから取得できます。定義上、イベントのイベント時間は処理時間よりも当然小さくなります。

ストリームコンピューティングエンジンの完全性に対する推論能力は、次のように簡潔に記述できます。システムは完全性信号を生成でき、その信号を何らかの方法でデータフロートポロジー全体にブロードキャストできる必要があります。トポロジー内の各オペレーターは、この信号に基づいて自身のデータ処理進捗を同期する必要があります。ストリームコンピューティングの入力集合を E と定義し、時刻 t（イベント処理）以降の入力集合を E(t) とすると、これには処理中のデータとバッファーされたデータが含まれます。この時のエンジン状態を State(t) とします。State(t) には各オペレーターの状態、データソースの消費オフセット（またはファイル読み取りオフセットなど）などが含まれます。ET 関数はイベント要素のイベント時間（またはその他の属性）を取得する関数を表します。オペレーターがイベントを処理する際、完全性とは、任意のイベント e ∈ E(t) に対して以下が成り立つことを意味します。

Signal(t) < ET(e)

オペレーターはこの信号とイベントの論理時間（すなわちイベント時間）を組み合わせて、現在の入力集合 E(t) のデータが完全かどうかを推論できます。一般的に、完全性信号は「(t-n, t) 区間の全イベントの最小イベント時間」として表現できます。この最小値は、何らかのアルゴリズム F により現在のデータソース入力とエンジン状態などを組み合わせて計算でき、以下のようになります。

Signal(t) = Min{ET(e) | e ∈ E(t) - E(t-n)} = F(E(t), State(t))

理想的には、入力集合 E(t) が t に伴い変化する全過程で上記の完全性制約が満たされるべきですが、この非常に強い保証は実際のシナリオ（すなわちアルゴリズム F）では実現が困難です。そのため、一般的にはいくつかの制約（時間など）を緩和し、一定期間内の完全性推論を得ます。

上記の完全性信号生成アルゴリズムの実装において、① は一定期間内のイベント時間をソートした最小値をセマフォとすることを意味します。② は過去の期間内の最小イベント時間をセマフォとしてカウントできることを意味します。③ はイベント時間から固定値を差し引いて完全性のセマフォとできることを意味します。本稿の第 4 パートで、上記 3 つの形式的な一般化表現が、リオーダリング、ローウォーターマーク、猶予時間という 3 つの工学的な完全性推論スキームに対応していることがわかります。

3.2 完全性推論のシステム設計

システムアーキテクチャの観点から見ると、ストリームコンピューティングエンジンがデータ完全性推論を実装する際、3 つの必須モジュールがあります。生成、伝播、消費です。生成モジュールは完全性信号を生成するために使用されます。シンプルなヒューリスティックアルゴリズムや適応型の複雑なアルゴリズムが存在します。これらのアルゴリズムは主に入力ソース自体の指標を組み合わせて使用します。たとえば、入力イベント内のイベント、ソース消費オフセット、データソースの上流側生産状態などです。生成モジュールは完全性推論の中で最も複雑な部分です。異なるエンジンは、パフォーマンス、複雑さ、ユーザーエクスペリエンスなどの面で異なる設計とトレードオフを行います。伝播モジュールは、完全性信号が生成されてからデータフロートポロジー全体にブロードキャストされるまでのプロセスです。このプロセスは、入力ソースに特殊要素を注入することで実現される場合も、ストリーム要素自体が持つ何らかの機能による場合も、データフロートポロジーの外部から各オペレーターに直接信号を送信する場合もあります。完全性信号の消費プロセスは比較的シンプルです。オペレーターは信号を受信した後、一般的に計算ウィンドウを閉じるか状態を排除するために使用されます。

3.3 完全性推論の工学的実装

業界におけるストリームデータの完全性推論は、イベントの再ソートが必要かどうかに基づいて大まかに 2 つのカテゴリーに分けられます。1 つはストリーム要素のバッファリングとリオーダリングを行うインオーダー処理（IOP）ストリームシステムと呼ばれます。IOP システムの代表的なものには Trill（Microsoft オープンソース）、Spark Streaming（D-Streams）、Aurora などがあります。もう 1 つはアウトオブオーダー処理（OOP）システムと呼ばれます。このタイプのシステムはデータをバッファリングして強制ソートを行わず、信号を通じてデータフローの処理進捗を追跡します。本稿で議論する完全性推論の内容は主にこの種のシステムを対象としており、MillWheel、Flink、Kafka Streams などが含まれます。

IOP システムでは、主にバッファリングとリオーダリングを通じてシステムに予測可能な完全性セマンティクスを提供できます。各イベントの到着により、それ以前のイベントが来ないことが保証され、これにより順序システムの設計が大幅に簡素化されます。リオーダリングスキームの最大の問題は、計算遅延を大幅に増加させることです。非順序の事前時間や空間制限を取得するのが通常困難であり、リオーダリングはストレージにも圧力をもたらし、総和、平均、カウントなど順序に依存しないオペレーター動作が必要であり、厳密な順序に依存できず、リオーダリングはこの種のオペレーターの特性を犠牲にします。

OOP システムでは、弱い属性概念を使用して完全性推論を実現します。主なスキームには、パンクチュエーション、ローウォーターマーク、猶予時間、ハートビートなどがあります。パンクチュエーションはデータフロートポロジーを通じて情報を転送する汎用メカニズムです。このスキームの考え方は、フローデータにいくつかの特殊な識別子を追加してデータ区間を区切ることです。これにより、非有界かつ非順序なフローデータを論理的に複数の有限データセットに分割できます。パンクチュエーションスキームの原論文は 2003 年に発表されました。パンクチュエーションスキームの設計が汎用的すぎて、工学的実践においてコストが非常に高いため、その後の多くのエンジンがパンクチュエーションスキームのアイデアを借用し、アーキテクチャの複雑さ、ユーザーエクスペリエンス、機能の完全性などの面でより合理的なスキームを設計しました。ローウォーターマーク、猶予時間、ハートビートなどがそれにあたります。以下の図はこれら 3 つのスキームのいくつかの類似点と相違点を示しています。

猶予時間はシンプルな完全性測定メカニズムです。一般的に、ストリーム要素のイベント時間から固定の時間長（要素がオペレーターに到着する際の最長遅延時間）を引いて猶予時間を算出します。この固定時間は、オペレーターの実際の処理サイクル（ウィンドウサイズの 2 倍など）に基づいてユーザーが設定できます。猶予時間スキームはフローに特別なメッセージを注入する必要がなく、実装は比較的シンプルですが、その欠点は完全性推論能力が不十分であることです。

ローウォーターマーク。ローウォーターマークはデータストリームに埋め込まれた特殊メッセージですが、通常のメッセージとは異なり、一般的に「ストリーム内に出現しうる最小タイムスタンプ」として表現されます。ローウォーターマークの考え方は、オペレーターがウォーターマークを受信すると、ストリームに関する追加情報を得るというものです。ウォーターマーク時刻より遅いデータは到着しないため、ウィンドウ内のデータを計算して下流に出力できます。ローウォーターマークの生成には、ヒューリスティックアルゴリズム（Kafka の全消費パーティションの最小オフセットの統計など）と、いくつかの適応型アルゴリズム（データソースの特徴の組み合わせなど）があります。

ハートビート検出。ハートビート検出は、データストリームの進捗情報を持つ外部信号で構成されます。信号にはタイムスタンプが含まれ、ハートビートタイムスタンプより大きいストリーム要素が「完全なデータセット」を構成します。ハートビート検出信号は入力ソースで生成されるほか、システムが環境パラメータ（ネットワーク遅延、入力ソース間の時計オフセットなど）を観察して推論することもできます。ハートビート検出スキームの利点は、ユーザーから隠されたエンジンの内部メカニズムであることです。ローウォーターマークスキームと同様、異なるデータソースの特徴に応じて完全性推論に理想的なハートビート検出信号を生成することは課題です。

上記のスキームの中で、IOP システムに適用可能なリオーダリングスキームのほかに、パンクチュエーション、ローウォーターマーク、猶予時間、ハートビート検出の 4 つのスキーム間には微妙な関連性と差異があります。ローウォーターマーク、猶予時間、ハートビート検出はすべてパンクチュエーションの特殊ケースと見なせます。猶予時間は信号伝播の経路を簡素化しています。フロー要素自体が持つ情報は猶予期間と組み合わせて計算できます。ハートビート検出はデータソースから得た完全性信号を直接エンジン入口（取り込みポイント）に送信します。ハートビート検出とローウォーターマークはどちらもデータソースに関する情報を伝達しますが、異なる点は、ローウォーターマークがデータソースから出力ソースのトポロジー全体に「完全性」情報を伝達するのに対し、ハートビート検出信号はデータソースの進捗情報のみを含むことです。

3.4 完全性推論と計算遅延

ストリーム処理の考え方はデータをオンラインで処理することであり、低い処理レイテンシが極めて重要です。強力な完全性を達成すると同時に、遅延は避けられません。遅延の理由は主に 2 つの部分から構成されます。1 つは完全性を達成するための「待ち」です。たとえば、ローウォーターマークスキームを使用するエンジンでは、集約データはローウォーターマークがウォーターマーク境界を越えた時にのみ下流に送信されます。2 つ目は、一部のエンジン（Cloud DataFlow など）がフォールトトレランスのために完全性信号を永続化することで、リンク全体の処理遅延が増加することです。

下流が中間結果を受け入れられる集約結果の計算では、完全性と遅延のバランスを取るソリューションとして、周期的なトリガーを導入することが挙げられます。実体化されたウィンドウ内のデータを反復的に（最終的には一貫して）更新し、データベースの世界でマテリアライズドビューを使用する場合に得られるセマンティクスに似たものを得ることで、下流は完全性を達成しながらタイムリーに最新の計算結果を取得できます。ただし、下流が中間結果を受け入れられない計算では、1 つの結果しか出力できません。遅延データに対して、エンジンは無限に待ち続けることはできません。エンジンが完全性信号を受信して結果を出力した後、遅延データがあればどう処理すべきでしょうか。一般的な処理戦略には、破棄、二重カウント、バイパス処理があります。二重カウントはより良い精度を達成できますが、長時間の状態保持が必要です。処理遅延とストレージコストを考慮すると、限られた度の「遅延猶予」しか実現できません。つまり、現在完全性推論を実現するスキームは多数ありますが、処理遅延などの要因を考慮すると、まだ長い道のりがあることがわかります。

4. ストリームデータ完全性のエンジン実装

現在、多くのストリームコンピューティングエンジンがストリームデータ完全性推論にローウォーターマークを使用しています。たとえば、Apache Flink と Google DataFlow はウォーターマークを API に配置し、ユーザーが直接使用できるようにしています。有界データや順序付きデータの処理に比べ、非順序データストリームでの理解とコーディングには一定のハードルがあります。収束（ユーザー使用）の複雑さの観点から、Kafka Streams と Apache Spark Structured Streaming はローウォーターマークの弱化版（猶予時間）を使用して最終的な一貫性結果を達成するか、遅延データによる状態ストレージコストを管理しています。次に、前述の章で挙げた完全性推論スキームと業界で最も成熟したストリームコンピューティングエンジンを組み合わせて、異なるエンジンが完全性推論の機能完全性、使いやすさ、複雑さなどのアーキテクチャ要素に関する考えとトレードオフを分析します。これらのエンジンから、非順序データ管理における多くのエンジンの開発トレンドをおおむね見ることができます。

4.1 MillWheel

MillWheel について説明する前に、MillWheel / Cloud DataFlow / Beam の関係を簡単に説明します。MillWheel は Google Cloud DataFlow の基盤ストリームコンピューティングエンジンです（現在、Google は内部的に Windmill に徐々に移行しています）。データ一貫性と完全性推論の問題を解決し、堅牢なストリームデータ処理を実現できます。Cloud DataFlow の最も顕著な貢献は、バッチ処理とストリームデータ処理に統一されたモデルを提供したことです。Beam は主にプログラミングモデル、汎用 API 層、ポータブル層で構成されています。最下層には具体的な実行エンジンがなく、リレーショナルデータベースにおける SQL に似て、ストリームバッチデータ処理の標準言語になろうとしています。

完全性推論プロセス：MillWheel / Cloud DataFlow はローウォーターマークのコアアイデアを採用し、イベント時間処理セマンティクスをサポートする完全性推論を実現しています。生産段階では、MillWheel / Cloud DataFlow データフロートポロジー内の各オペレーターの物理ノードが、当該ノード上のデータ処理進捗（処理済み + 処理待ち）を追跡します。進捗情報をメモリとローカル状態に永続化すると同時に、定期的に中央機関に進捗情報を送信します。中央機関での定義は、当該オペレーターに対応する全ノードが報告した処理進捗の最小値です。伝播段階では、集約システムが計算した最終ローウォーターマーク情報が対応するオペレーターの物理ノードに送信され、ノードの最終ウォーターマーク = Min（現在のノード処理進捗、上流グローバルウォーターマークの最小値）となります。消費段階では、MillWheel / Cloud DataFlow のノードが計算されたローウォーターマーク情報を使用して、事前に暗黙的（ウィンドウトリガーなど）または明示的（プログラミング）に登録されたタイマーを走査し、ローウォーターマークより小さいタイムスタンプのタイマーコールバックを実行して、結果の送信や状態の変更を完了します。

アーキテクチャ設計のトレードオフ：MillWheel / Cloud DataFlow はノードでローウォーターマークを永続化する必要があり、ウォーターマークの更新には集中報告が必要です。そのため、これら 2 つの部分の IO オーバーヘッドがストリーム処理遅延の増加を引き起こします。集中報告モードでは、Cloud DataFlow のオペレーターは動的にパーティショニングされており、データソースは非常に複雑です。完全性推論を保証するためには、より複雑なローウォーターマーク生成メカニズムが必要です。ノード永続化とローウォーターマークは遅延をもたらすと同時に、1 つの利点をもたらします。MillWheel / Cloud DataFlow アプリケーションのフェールオーバーが高速であることです（より細かい状態粒度を持ちます）。同時に、グローバルウォーターマーク集約はエンジンの現在のデータ処理進捗をより良く評価できます。たとえば、計算遅延と精度のトレードオフとして 98% ローウォーターマークを計算でき、これはシステム内のストリームデータの 98% の処理進捗に対応し、全体の処理効率をより速くできます。

4.2 Apache Flink

Apache Flink の完全性推論スキームは、DataFlow モデルとローウォーターマークに基づいて設計されています。

完全性推論プロセス：生産段階では、Flink プログラムはソースノードまたは専用ウォーターマーク生成ノードでウォーターマークを生成できます。ソースノードはエンジンに入るストリームデータ情報またはデータソースのその他の情報（Kafka パーティション、オフセット、タイムスタンプなど）に基づいてウォーターマークを計算できます。一部の専用ウォーターマーク生成ノードは、観測するストリーム要素のタイムスタンプに基づいてウォーターマークを計算できます。伝播段階では、データフロートポロジー全体において、ウォーターマークは通常のストリームデータとともに特殊なメタデータメッセージとして下流ノードに送信されます。下流ノードは新しいウォーターマークメッセージを受信すると、すべての入力ウォーターマークの最小値を現在のノードのウォーターマークとして採用し、同時に現在のノードのウォーターマークを更新します。ノードは以前のウォーターマークより大きいウォーターマークのみを転送し、完全性信号の厳密な単調性を保証します。消費段階では、MillWheel / Cloud DataFlow と同様に、ウォーターマークがノードに到着すると一連のタイマーがトリガーされ、結果が下流に送信されます。新しいウォーターマーク値はすべての下流ノードにブロードキャストされ、分散アプリケーション全体で状態同期を実現します。

アーキテクチャ設計のトレードオフ：Flink では、ノードがローウォーターマークを永続化しないため、ノード障害時にパイプライン全体を一時停止し、最後のチェックポイントから復旧する必要があります。障害ノードのローウォーターマークは低い値の定数として設定されます。ノードがすべての入力エッジで新しいローウォーターマークメッセージを受信すると、ウォーターマークがリセットされます。ローウォーターマークを永続化しない利点は、Flink のエンドツーエンドデータ処理遅延が低いことであり、欠点はフェールオーバーに MillWheel / Cloud DataFlow より時間がかかることです。Flink は完全性推論の設計において最も成熟したエンジンです。

4.3 Apache Kafka Streams

Cloud DataFlow と Flink はどちらもローウォーターマークを使用してストリームデータ完全性推論を行っていますが、Apache Kafka Streams のエンジニアはローウォーターマークスキームに基づく完全性推論メカニズムを構築しませんでした。主な理由は以下の 2 つの側面です。第 1 に、Apache Kafka Streams は「継続的インクリメンタル処理ストリームテーブル」モデル（DataFlow のストリームテーブル概念に類似、「Streaming System」を参照）を提案しており、テーブル内の結果がフローを通じて段階的に更新され中間結果も同時に発行される場合、ウィンドウをいつ閉じるかの概念があまり重要でなくなります（個人的には、この主張は実践的に問題があります。たとえば、下流で異なる結果の同一性を定義するのが難しく、遅延処理の累積と撤回のロジックの実装は一般的に複雑です）。第 2 に、エンジニアは DataFlow モデルのローウォーターマークスキームが複雑すぎると考えており（たとえば、少なくとも 8 つのトリガーをウォーターマークと組み合わせる必要がある）、ユーザーにより簡潔で直感的な完全性ソリューションを提供する必要があります。上記 2 点に基づき、Apache Kafka Streams は前述の猶予時間を使用して完全性問題を解決しています。

完全性推論プロセス：Apache Kafka Streams はストリームに特別なメタ情報を埋め込まず、システムレベルのローウォーターマークタイムスタンプにも依存しません。代わりに、各オペレーターで猶予期間を設定することで、きめ細かい完全性判断を可能にしています。生産段階では、各イベントがオペレーターを通過する際、オペレーターは「イベント時間 - 猶予時間」を完全性信号として使用します。ストリームデータが継続的に流入するにつれてイベント時間も増加するため、「ローウォーターマーク」に似た進捗推論が得られます。伝播段階では、信号はイベント自体が持つ情報から計算できるため、信号伝播プロセスは存在しません。もちろん、いくつかの問題があります。たとえば、上流のオペレーターが大量のデータをフィルタリングした場合、下流のオペレーターは長時間データを受信しないため、現在のオペレーターのストリーム処理進捗をタイムリーに進められない可能性があり、これは処理遅延をある程度増加させます。消費段階では、猶予時間で計算されたタイムスタンプがウィンドウの上限を超えると、ウィンドウが閉じられて状態が解放されます。

アーキテクチャ設計のトレードオフ：Cloud DataFlow と Flink のローウォーターマークスキームを一般化すると、ここでの猶予時間はウォーターマークスキームのサブセット、すなわち「ハイウォーターマーク」のような設計と見なせることがわかります。猶予時間スキームの利点は、ローウォーターマークの概念をグローバルレベルからオペレーターレベルに拡張したことであり、異なるオペレーターの待ち時間をデカップリングできます。たとえば、6 時間の集約ウィンドウと 10 分間の集約では、一般的に非順序データに対する待ち時間が異なります。異なる猶予時間の設定を通じて、より柔軟な同期制御を実現できます。猶予時間の欠点は主に「信号が現在のイベント情報を使用する」ことに起因します。前述の下流フィルターオペレーターが処理進捗をタイムリーに取得できないという欠点に加え、データフロートポロジーにグローバル同期機能が欠けており、この推論は一部のシナリオで不正確な結果を引き起こす可能性があります。たとえば、集約オペレーターの上流のあるノードの処理進捗が GC やホストの負荷により遅延しているが、他のノードは正常に処理しているとします。集約オペレーターが受信するイベント時間は徐々に増加しているため、猶予期間が経過するとウィンドウは閉じます。遅延ノードが処理進捗に追いついた時、下流の集約オペレーターに送信されるデータはすでに使用できません（ウィンドウは閉じられています）。上記の問題を解決するため、Apache Kafka Streams は DataFlow モデルに似たアプローチを使用して、進捗情報を持つメタデータをストリームデータに注入することで進捗追跡を実装する予定です。ただし、このメタデータは必ずしもシンプルなタイムスタンプではなく、Vector Clock のようなソリューションで、きめ細かいオペレーター制御を基盤にグローバルなデータフローグラフ同期を実現する可能性があります。

4.4 Apache Spark Streaming

Spark のフロー処理はバッチ処理設計から派生しているため、エンドツーエンドの一貫性とタイムイベントのサポートは十分ではありません。Spark 2.1 以降、新しい Apache Spark Structured Streaming API（SPARK-18124）は、猶予時間ベースのウォーターマークに似たデータ完全性スキームを導入しました。ユーザーはイベント時間列と遅延データの猶予時間を指定することで遅延データを管理でき、エンジンはフロー状態のメモリ使用量を制御できます。遅延イベントの破棄や決して更新されない古い状態（集約や接続など）の削除などが可能です。

完全性推論プロセス：Apache Spark Structured Streaming のウォーターマークはグローバルです。各バッチ計算のトリガー後にウォーターマークは再計算されます。具体的な計算ロジックは、新しいウォーターマーク = MAX（トリガー実行前に確認された最大タイムスタンプ、トリガー実行データ内の最大タイムスタンプ） - 猶予時間です。複数の入力ソースがある場景では、Apache Spark Structured Streaming は各入力ストリームを追跡し、個別にウォーターマークを計算し、最小値をグローバルウォーターマークとして選択します。グローバルウォーターマークに基づき、Apache Spark Structured Streaming は到着データの狀態を維持し、遅延データと集約して更新できます。ウォーターマークより小さい遅延データは集約され、ウォーターマークを超えるデータは破棄されます。

アーキテクチャ設計のトレードオフ：Apache Spark Structured Streaming のウォーターマークは計算における状態管理を目的として設計されているため、その完全性推論能力は比較的弱いです。たとえば、グローバルウォーターマークを使用しているため、データフロートポロジー内で連鎖的な集約を実行できません（下流の集約オペレーターの入力が上流の集約オペレーターの出力である場合）。これは不正確な集約結果を招く可能性があります。さらに、ウォーターマーク API の設計では、適用範囲にも制限があります。たとえば、ウォーターマークの時間列は後続のグループ化列と一致する必要があり、これにより集約オペレーターの機能が大幅に制限されます。

4.5 エンジン完全性実装スキームの比較

次に、完全性推論信号の生成と伝播の観点から、前述の 4 つのストリームコンピューティングエンジンの実装を比較します。

5. まとめと展望

ストリームコンピューティングエンジンを通じて正しい結果を得るには、エンジン一貫性とデータ完全性の両方に協力して取り組む必要があります。エンジン一貫性の本質は分散アプリケーションのフォールトトレランス問題であり、現在、工学的実践において体系的で成熟したソリューションが存在します。各エンジンの開発方向性も似ており、すなわち「トランザクション方式で出力結果（状態）を出力する」ことです。データ完全性はストリームコンピューティングにおける非順序かつ非有界データの確定的なデータセットを保証するものであり、単一集約結果、欠損検出、インクリメンタル処理などのシナリオで極めて重要です。データ完全性推論をいかに実現するかは、ユーザーにとって最も理解が難しく、ストリームコンピューティング分野で完成度も十分ではない問題の 1 つです。完全性推論の本質は、完全性信号生成アルゴリズム Signal を持つことであり、任意のイベント e と処理時間 t に対して、Signal(t) < e のイベント時間が成り立つことです。現在の工学的実践において、完全性推論スキームにはリオーダリング、パンクチュエーション、ローウォーターマーク、猶予イベント、ハートビート検出などがあります。中でもローウォーターマークと猶予時間が最も広く使用されています。これら 2 つのスキームの多くはヒューリスティックであり、可能な限り正確に完全性推定を与えることしかできません。実際のコストと実際の需要を考慮すると、現在絶対的に正しい完全性推論アルゴリズムは存在しません。

すべてのストリームコンピューティングエンジンが正しい結果を得られるわけではありません。データ完全性を達成するためのさまざまな制約を理解することは、正しい技術選定にとって極めて重要です。MillWheel と Flink はローウォーターマークを使用して、データフロートポロジー全体の処理進捗を同期しています。ローウォーターマークは特殊なストリーム要素またはバイパス伝播方法を使用し、データフロートポロジー全体の各オペレーターが現在のデータ処理進捗を明確に把握できるようにします。Apache Spark Structured Streaming と Kafka Streams は猶予時間スキームを採用し、アーキテクチャ構築とメンテナンスの複雑さを簡素化し、ユーザーの完全性理解のコストを低減しています。同時に、これら 2 つのエンジンは完全性推論機能においていくつかの欠点も持っています。

絶対的に正しい完全性と遅延のパラドックスにおいて、ストリームコンピューティングエンジンも異なる方向から現在の技術ソリューションを最適化しようとしています。たとえば、データソースのベンダー（エンジンではなく）が完全なウォーターマークを送信することでエンジン設計コストを削減し、組み込みの適応型エンジンウォーターマークアルゴリズムでヒューリスティックアルゴリズムの不足を補っています。同時に、エンジン一貫性と比べて、データ完全性の概念はユーザーに近く、完全性の多くの定義にはユーザーの参加が必要です。私の見解では、現在の多くのエンジンの完全性ソリューションは、エンジンフレームワークのユーザーにとってまだハードルが高いと感じています。フレームワークユーザーに、正しい結果を達成するために必要な指標制約をよりフレンドリーな方法で表現させることが、解決すべき重要な命題です。これらの問題がうまく解決できれば、ストリーミングとバッチの真の統合と、バッチ処理を超えるストリームコンピューティングが現実のものとなる可能性があります。