New developments in Flink Runtime for streaming and batch integration

1. ストリームバッチ統合

ストリームバッチ統合の目的は、有界データと非有界データに対して統一された処理 API のセットを提供することである。
これには Datastream API と Table/SQL API が含まれる。
有界データの処理はオフライン処理に対応し、非有界データの処理はオンライン処理に対応する。


このようなストリームバッチ統合処理 API が必要とされる理由は主に 2 つある。


まず、リアルタイムコンピューティングの継続的な発展に伴い、企業のデータ処理パイプラインの多くがオフライン処理とオンライン処理で構成されている。
同一の開発 API を使用することで、ストリーム処理とバッチジョブ開発の学習コストとメンテナンスコストを削減できる。


また、多くのシナリオにおいて、ユーザーのストリーム処理ジョブは遅延データやオンラインロジックの変更に制約を受けることがある。
たとえば、ユーザーがコメントを送信するまでに時間がかかる場合や、オンライン処理ロジックのアップグレードが必要になる場合がある。
このような場合、オフラインジョブを使用して以前の処理結果を修正する必要がある。
つまり、バックフィルのケースである。


この場合、2 つの異なる API を使用していると処理結果の一貫性を維持することが困難になるため、ユーザーは上記の問題を解決するために統一 API を必要とする。


Flink のストリームバッチ API は Datastream API と Table/SQL API の 2 つで構成される。
Table/SQL API は比較的ハイレベルな API で、主に標準 SQL と同等のテーブル操作を提供する。
一方、Datastream API は比較的ローレベルで、ユーザーはオペレーターの時間とステートを明示的に操作できる。
これら 2 つの API セットは相互に変換して組み合わせて使用できる。


これら 2 つの API セットに対し、Flink は 2 つの異なる実行モードを提供する。


ストリーミング実行モードは、オペレーターのステートを保持し、新しいデータ到着時に増分計算を実行する方式である。
有界データセットと非有界データセットの両方の処理に使用でき、リトラクションなど任意の処理ロジックをサポートする。
すべての履歴データを保持し、新しいデータが到着すると保持された履歴データを更新し、過去に受信したすべてのデータを再順序付けし、最後に以前送信した結果に対してリトラクションを実行する。
たとえば、reduce オペレーターはさらに最適化して、無限の履歴ステートを実際に保存しないようにできる。
さらに、増分計算ではデータの到着順序が不定なため、SQL へのアクセスも順不同となり、ランダム IO が発生する可能性がある。
最後に、ストリーム処理モードはフェールオーバーをサポートするために定期的なチェックポイントに依存しており、これによる処理オーバーヘッドも生じる。


そのため、有界データセットの処理には、専用のバッチ処理モードも提供している。
オペレーターは段階的に処理されるため、有界データの処理にのみ使用できる。
この場合、オペレーターの実装では特定の最適化が可能である。
たとえば、データを先にソートし、キーごとに順番に処理することで、無限ステートやランダム IO の問題を回避できる。


Flink は、同一の有界入力データに対して 2 つの実行モードで処理結果が一致することを保証できる。
さらに、2 つの異なるモードに対して統一されたパイプラインリージョンスケジューラ、統一されたシャッフルサービスプラグインインターフェイス、統一されたコネクタインターフェイスを提供し、2 つのインターフェイスに統一されたサポートを提供している。


現在、Flink のアーキテクチャは上図の通りである。
API の面でも具体的な実装の面でも、ストリームとバッチの統合を全体として実現した状態となっている。


2. セマンティクスの強化と改善

上記のストリームバッチ統合の定義に対し、近年のバージョンでも継続的な改善と最適化を行っている。
最初の部分はストリームバッチ統合セマンティクスの強化と改善に関するものである。


まず、ストリーミングモードで一部のタスクが終了した後もチェックポイントを継続できるようにする機能をサポートした。


現在のプロセスにおけるジョブの終了は 2 つの状況に分けられる。


ソースが有界の場合、ジョブは最終的に実行完了する。


非有界ソースの場合、ユーザーは stop-with-savepoint--drain コマンドでジョブを終了し、セーブポイントを保持できる。
drain パラメーターを指定しない場合、drain 操作は実行されない。
この場合、セーブポイントは一般的にジョブの再起動のために保持され、ジョブの終了には該当しない。


以前の Flink は一部のタスク終了後のチェックポイントをサポートしていなかった。
これにより 2 つの問題が発生するためである。


第一に、2 フェーズコミットのシンクはチェックポイントに依存してストリーミングモードでのエンドツーエンドのデータ整合性を実現している。
この場合、2 フェーズコミットのシンクはまずデータを一時ファイルまたは外部トランザクションに書き込む。
Flink 内部のチェックポイントが成功した後、つまりチェックポイント以前のデータがリプレイされないという前提のもとで、2 フェーズコミットのシンクはファイルの名前変更またはトランザクションのコミットによって実際にトランザクションを安全にコミットできる。
一部のタスク完了後にチェックポイントができない場合、データの最後の部分を常にコミットできず、ストリームモードとバッチモードの処理結果の一貫性を保証できない。


第二に、有界データソースと非有界データソースの両方を含む混合ジョブの場合、一部の実行完了後にチェックポイントを実行できないと、その後の実行でフェールオーバーが発生した際にロールバックによる大きなオーバーヘッドが生じる。


上記の問題を解決するため、一部のタスク完了後にチェックポイントを実行する機能をサポートし、ジョブ完了のプロセスを変更して、2 フェーズコミットを使用するタスクが最後にチェックポイントの完了を待ってから終了できるようにする必要がある。
通常の終了の場合は次のチェックポイントの完了を待ってから終了し、drain の場合はセーブポイントの完了を待ってから終了する。


一部のタスク完了後にチェックポイントを実行できるようにするため、チェックポイントのプロセスを変更した。
まず、新しいソースタスク、つまり先行タスクは終了しているが自身はまだ終了していないタスクを再識別し、これらのタスクからバリアを送信して通常通りのチェックポイント操作を実行する。
チェックポイントのステートは JobVertex 単位で記録されるため、ある JobVertex のすべてのタスクが終了した場合はそのステートに特別なマークが記録される。
一部のタスクが終了した場合は、実行中のすべてのタスクのステートが JobVertex のステートとして保持される。
他のすべての JobVertex の処理フローは通常のチェックポイントと一致する。
フェールオーバー発生後にジョブが再起動されると、完全に終了した JobVertex はスキップされ、他のタスクの処理ロジックは通常の処理ロジックと一致する。


上記の作業に基づき、ジョブ完了後のプロセスとセマンティクスも再整理した。
2 フェーズコミットのシンクが最後のデータ部分をコミットしてから終了できるようにするため、これらのタスクは最後のチェックポイントを待ってから終了する必要がある。
現在のロジックでは、ジョブが自然終了する際、まず max ウォーターマークを送信し、次に EndOfPartitionEvent を送信する。
タスクが EndOfPartitionEvent を受信すると、それぞれオペレーターの endOfInput()、close()、dispose() 操作を呼び出す。
最後にチェックポイントを挿入する場合、最適な方法は close メソッドの後に挿入することである。
ここでタスクはすべての作業を完了しているからである。


ただし、実際のシナリオでは違いがある。
実際のシナリオではセーブポイントがトリガーされる。
セーブポイント成功後、ソースは終了メソッドを呼び出して実行を終了し、max ウォーターマークと EndOfPartitionEvent を送信する。
後続のロジックはチェックポイントの場合と一致する。
以前にセーブポイントが実行されているため、close の後にチェックポイントを実行すると冗長になる。
この場合、より適切な方法は、まずジョブを終了し、その後セーブポイント操作を実行し、セーブポイント操作と同時にデータの最後の部分をコミットすることである。


ただし、ここにも問題がある。
最後のセーブポイントを保持する場合、すべてのタスクが同じセーブポイントの終了を待つ必要がある。
自然終了の場合、異なるタスクが異なるチェックポイントを待って終了できる。
しかしセーブポイントの場合、ジョブ終了前に EndOfPartitionEvent が送信されており、これによりタスク間のネットワーク通信が閉じられるため、ジョブ終了後に元のソースからセーブポイントを実行できなくなる。


これらの問題を解決するため、ネットワーク接続を閉じずにすべてのタスクに終了を通知し、すべてのタスクが終了した後にセーブポイント操作を開始し、成功後にネットワーク接続を閉じる必要がある。
これにより、すべてのタスクが終了時に同じセーブポイントのステートを待ってから終了できる。


この変更をサポートするため、新しい EndOfDataEvent を導入した。
タスクが EndOfDataEvent を受信すると、以前の EndOfPartitionEvent の処理を呼び出す。
その後、ソースはすぐにバリアを送信してセーブポイント操作をトリガーし、オペレーターは終了後に終了ロジックを実行する。


さらに、以前あいまいだった close() と dispose() 操作をそれぞれ finish() と close() に改名した。
finish() はタスクが正常に終了した場合にのみ呼び出され、close() はジョブの正常終了時と異常終了時の両方で呼び出される。


セマンティクスの部分でもう一つ実施した作業はハイブリッドソースである。


ハイブリッドソースは、履歴バッチデータの読み取り後に有界ストリームデータに切り替えて処理することをサポートする。
処理ロジックが一致している条件下的でのストリームバッチ移行操作に適している。
つまり、リアルタイムデータが既にディスクに配置されており、ユーザーがバックフィルを実行する必要がある場合で、ストリームバッチ処理ロジックが一致している場合、ユーザーはハイブリッドソースを使用してジョブを容易に実装できる。


3. パフォーマンス最適化

セマンティクスに関する作業に加え、ランタイムレイヤーでもいくつかのパフォーマンス最適化を実施した。


3.1 スケジューリングデプロイパフォーマンスの最適化

まず、スケジューリング部分のパフォーマンス最適化について述べる。
Flink のオールツーオール接続により、同時実行数 n の 2 つのオペレーター間に n² のエッジが存在する。
これらの n² のエッジは JM のメモリに明示的に保存され、多くのスケジューリングとデプロイのロジックも直接これに依存して処理を行うため、JM のメモリ空間と多くの計算の時間計算量および空間計算量が O(n²) となる。
バッチジョブは一般的にサイズが大きく、スケジューリングがより細かい粒度であるため、スケジューリングとデプロイのパフォーマンスの問題を悪化させる。


この問題を解決するため、オールツーオールエッジの対称性を利用してメモリ内のデータ構造と計算ロジックを再構築し、コンシューマーグループのデータ構造を導入して、以前の ExecutionEdge に代わってオペレーター間の接続関係を統一的に記述する。
この方法では対称性の情報を繰り返し記述しないため、n² の複雑さを回避できる。
この新しい記述に基づき、メモリ内で ExecutionEdge を維持しなくなった。


さらに、パイプラインリージョンの計算、タスク終了後にスケジューリングが必要な後続タスクの計算など、多くのスケジューリングアルゴリズムを調整し、それらの時間計算量を O(n) に削減した。


パイプラインリージョンの計算プロセスには特別なロジックもある。
Flink のジョブ DAG グラフには 2 種類のエッジが含まれる。
パイプラインエッジとブロッキングエッジである。
前者は上流と下流のタスクを同時に起動し、ネットワーク経由でデータを送信する必要がある。
後者は上流と下流のタスクを順次起動し、ファイル経由でデータを送信する必要がある。
スケジューリング前にまずパイプラインリージョンを計算する必要がある。
一般的にはブロッキングエッジで分断し、パイプラインエッジで接続されたすべてのタスクを同じリージョンに配置するが、上図に示すようにこのロジックには問題がある。
同時実行 1 のタスクと同時実行 2 のタスクがブロッキングエッジを通じて 2 つのリージョンに分割される場合、ブロッキングエッジで直接分断すると 2 つのリージョンに分割される。
そして task1 と task2 の間にオールツーオールのシャッフル関係があるため、リージョンで構成されるグラフ上に循環依存の関係が発生し、スケジューリング時にデッドロックが生じる。


以前のプラクティスでは、Tarjan 強連結成分アルゴリズムを使用してこの循環依存を識別していた。
当時、識別は実行グラフ上で直接行われていたため、オールツーオールエッジに n² の接続が存在することから時間計算量は O(n²) であった。
さらに分析した結果、パイプラインの検証と識別をジョブグラフ上で直接実行すれば、グラフ内にオールツーオールエッジが存在する限り必ず循環依存が発生するため、ジョブグラフ上で直接判定してすべてのオールツーオールエッジを識別できることが分かった。
その後、ExecutionGraph 上で非オールツーオールエッジを処理する。


この方法で、循環依存識別の複雑さを O(n) に削減できる。


3.2 デプロイパフォーマンスの最適化

もう一つの最適化はデプロイパフォーマンスに関するものである。


Flink はタスクのデプロイ時にシャッフル記述子を含める。
上流の場合、シャッフル記述子はデータがどこで生成されるかを記述し、下流の場合はデータをどこからプルする必要があるかを記述する。
シャッフル記述子の数は ExecutionEdge と等しく、このオーダーは O(n²) である。
メモリ内での計算、シリアライズ、保存時にシャッフル記述子は大量の CPU とメモリを消費し、メインスレッドがブロックされ、TM のメモリ枯渇問題を引き起こす。


しかし、上流と下流の対称性により、多くのシャッフル記述子は実際には重複している。
シャッフル記述子をキャッシュすることで維持数を削減できる。


さらに、同時実行数の過多によりシャッフル記述子が大きくなりすぎてメモリ OOM が発生するのを防ぐため、BlobServer を使用してシャッフル記述子を転送するようにした。


上記の最適化を実現した後、10000×10000 のオールツーオールの 2 段ジョブを使用してテストを行った。
その結果、スケジューリングとメモリ使用量が 90% 以上削減され、デプロイ時間が 65% 以上削減され、スケジューリングとデプロイのパフォーマンスが大幅に向上したことが確認できた。


ストリーム実行モードのスケジューリングとデプロイの最適化により、ジョブのフェールオーバー時の再起動時間が大幅に短縮されたが、フェールオーバーが発生すると再デプロイ、初期化、ステートの読み込みに依然として一定の時間がかかる。
この時間をさらに短縮するため、ジョブのフェールオーバー時に失敗したノードのみを再起動することを試みている。
難しさはデータの一貫性をどのように保証するかにあり、現在検討中である。


ランタイム最適化のもう一つの部分は、ストリーミングモードで Buffer Debloating を通じてバッファーサイズを動的に調整し、バックプレッシャー下でのチェックポイントのバッファーアラインメントに必要な時間を短縮してチェックポイントのタイムアウトを回避することである。
バックプレッシャーが発生し、ジョブの中間にキャッシュされるデータ量が大きすぎる場合、バッファーのサイズをタイムリーに縮小して中間キャッシュのデータサイズを制御し、データ処理によるバリアのブロックを回避できる。


4. リモートシャッフル

シャッフルはバッチジョブ実行において非常に重要な部分である。
クラウドネイティブは統一された運用保守 API を提供し、運用保守オーバーヘッドを削減し、オフライン混合や動的スケーリングの場合により良いサポートを提供できるため、近年 Flink はクラウドネイティブを積極的に取り入れている。
このバージョンでは Kubernetes 上の Flink の完全な実装や動的スケーリングをサポートするスケジュールなど、クラウドネイティブを積極的に採用している。
しかし、Flink のシャッフルはローカルディスクを使用する必要があるため、クラウドネイティブな Flink をサポートするにはストレージとコンピューティングを分離したシャッフルを実装する必要がある。
ストレージとコンピューティングを分離するアーキテクチャにより、コンピューティングリソースとストレージリソースを独立してスケーリングでき、コンピューティング完了後にタスクマネージャーがすぐに終了するのを防ぎ、リソース全体の利用率を向上できる。
同時に、タスク実行の失敗による TM の終了がシャッフルファイルサービスの安定性に影響を与え、下流の実行に影響するのを防ぐこともできる。

上記のシャッフルのストレージとコンピューティングを分離する要件に対応するため、社内でリモートシャッフルサービスも開発した。この機能は今年初めに社内でリリースされ、試用期間を経て、少し前にこのシステムをオープンソース化した。以下でこのシステムについて詳しく紹介する。

Flink はさまざまなシナリオをサポートでき、異なるシナリオでのシャッフルは記憶媒体、転送、デプロイ方法において大きく異なる。たとえば、ストリーム処理モードでは、Flink は一般的にネットワークベースのオンライン転送方式を採用する。データは上流 TM のメモリにキャッシュされ、下流タスクに空きバッファーがあるときにタイムリーに送信される。分析処理モードでは、オペレーターの段階的な操作をサポートするため、Flink はファイルベースのオフライン転送方式もサポートする必要がある。これはまずオフラインファイルに書き込み、下流タスクの起動後にネットワークを通じて下流タスクに送信される。オフラインファイルはローカル TM に存在することも、リモートサービスに存在することもできる。

さらに、異なるシャッフルはライフサイクル管理、メタデータ管理、データ分散戦略においても多くの共通要件を持つ。すべてのシャッフルは、スケジューラが上流タスクの起動時に対応するシャッフルリソースを申請し記録する必要がある。また、スケジューラが下流タスクのデプロイ時にシャッフルのリソース記述子を含め、下流タスクが対応するデータをスムーズに読み取れるようにする必要がある。最後に、シャッフルは終了や実行失敗などの特定のライフサイクル時にスケジューラに依存してリソースをクリーンアップする。

異なるシャッフルに統一されたサポートを提供するため、Flink はバージョン 1.9 からプラグインシャッフルアーキテクチャを導入した。シャッフルプラグインは主に 2 つの部分で構成される。シャッフルマスターは JM 側のスケジューラとのやり取りを担当し、シャッフルリソースの申請と解放の機能を実現する。Result Partition と Input Gate はそれぞれデータの書き込み端と読み取り端として使用され、スケジューラが提供するシャッフルリソース記述子に従ってデータを特定の場所に出力または読み取る。すべてのシャッフル実装において、共通部分は Flink によって実装される。スケジューラはパーティショントラックを通じて申請されたシャッフルリソースを記録し、ジョブの実行モードに従ってシャッフルリソースのライフサイクルを維持する。

統一されたプラグインシャッフルインターフェイスを通じて、Flink は異なるシャッフル実装の複雑さを簡素化し、異なるシャッフルが実際のストレージと転送方法を自由に選択できるようにする。

Flink の統一されたプラグインシャッフルインターフェイスに基づく、Flink リモートシャッフルの全体的なアーキテクチャを上図に示す。シャッフルサービスは独立したクラスターによって提供され、シャッフルマネージャーはクラスター全体のマスターノードとして機能し、ワーカーノードの管理、およびシャッフルデータセットの割り当てと管理を担当する。シャッフルワーカーはクラスター全体のスレーブノードとして、データセットの読み取り、書き込み、クリーニングを担当する。シャッフルマネージャーはハートビートを通じてシャッフルワーカーとシャッフルマスターを監視し、ハートビートタイムアウト時にデータの削除と同期を行い、クラスター内のデータセットのステートの一貫性を保つ。

転送プロセスにも多くの最適化を施した。ネットワーク部分はクレジットベースプロトコルに基づいて実装されており、Flink の既存のネットワーク転送プロトコルと類似している。TCP 接続の多重化、圧縮、制御可能なメモリ管理、ゼロコピーなどの一連の最適化も実装した。IO 部分については、IO スケジューリングの最適化をサポートする mapPartition ストレージフォーマットを提供した。IO スケジューリングの最適化により、HDD 上でのアクセス速度は 140 MB/s 以上に達する。

さらに、現在プリマージに基づく reducePartition ストレージフォーマットを開発中である。これは下流先に応じてデータを事前にマージし、特定のワーカー上に保存する。下流がすべて同時に起動できない場合、mapPartition よりも優れた結果を得られる。

デプロイ面では、リモートシャッフルはさまざまなデプロイ方法をサポートできる。さらに、バージョン間のプロトコル互換性も提供し、サーバーのアップグレード時にクライアントをアップグレードする必要がないようにしている。最後に、システムに共通のメトリック操作も提供しており、追加の運用保守ツールも積極的に開発中である。

5. まとめと今後の展望

全体的に、Apache Flink は現在、オンラインに投入可能なストリームバッチ統合のデータ処理能力を備えている。今後、この能力をさらに改善し、より多くのストリームバッチ統合シナリオでのサポートを提供する。たとえば、ハイブリッドソースとは異なり、バックフィルのシナリオでストリームバッチ処理のロジックが一致しない場合、バッチ処理終了後にステートを保持してストリーム処理ジョブを開始する方法のサポートも検討している。

さらに、システム全体の安定性とパフォーマンスをさらに改善し、ストリームモードとバッチモードの本質的な違い、およびストリームバッチ統合のより深い意味についてさらに深く考察していく。