Efficient and Stable Universal Incremental Checkpoint

I. 概要

Generic Log-Based Incremental Checkpointing の元の設計意図は、完全な状態スナップショットを増分チェックポイント機構から分離し、増分 Changelog を継続的にアップロードすることで各チェックポイント処理を安定かつ迅速に完了できるようにし、それによってチェックポイント処理間の間隔を短縮して Flink システムのエンドツーエンド遅延を低減することです。拡張性に関しては、主に以下の 3 つの改善があります。

より短いエンドツーエンドレイテンシ：特に Transactional Sink において。Transactional Sink は Checkpoint 完了時にのみ 2 フェーズコミットを完了できるため、チェックポイント間隔を短縮することはより頻繁なコミットと短いエンドツーエンド遅延を意味します。

より安定したチェックポイント完了時間：現在、チェックポイント完了時間は主にチェックポイント処理中に永続化が必要な（増分）状態のサイズに大きく依存しています。新しい設計では、増分を継続的にアップロードして Checkpoint Flush 中に永続化が必要なデータを減らすことで、Checkpoint 完了の安定性を確保しています。

フォールトトレラントリカバリ時にロールバックが必要なデータがより少ない：チェックポイント処理間の間隔が短いほど、各フォールトトレラントリカバリ後に再処理が必要なデータは少なくなります。

これはどのように実現されるのでしょうか。Flink のチェックポイント時間に影響する主な要因は以下の通りです。

Checkpoint Barrier のフローとアライメントの速度。
状態スナップショットを不揮発性の高可用性ストレージ（S3 など）に永続化するために必要な時間。
Flink Checkpoint 機構に詳しくない読者は、1 を参照してください。

Flink バージョン 1.12 で導入された Unaligned Checkpoint[2] とバージョン 1.14 で導入された Buffer Debloating[3] は、主に上記の最初の問題、特にバックプレッシャーが発生している場合に対処します。以前に導入された Incremental Checkpoint [4] は、各チェックポイント処理に必要な永続ストレージ状態のサイズを削減して 2 番目の影響要因を軽減するものですが、実際には完全には可能ではありません。既存の Incremental Checkpoint は RocksDB ベースで行われ、RocksDB はスペース拡大と読み取りパフォーマンスの考慮から定期的にコンパクションを実行します。コンパクションは新しい比較的大きなファイルを生成し、アップロードに必要な時間を増加させます。Flink ジョブを実行する各物理ノード（タスク）には少なくとも 1 つの RocksDB インスタンスがあるため、物理ノード数が増えるにつれてチェックポイントが遅延する確率が増加します。その結果、Flink の大規模ジョブでは、ほぼ毎回チェックポイント処理完了時に特定のノードによって遅延する可能性があり、下の図のようになります。

また、既存のチェックポイント機構では、タスクは少なくとも 1 つの Checkpoint Barrier を受信した後にのみ状態スナップショットを取得し、高可用性ストレージへの状態スナップショットの永続化を開始するため、チェックポイント完了時間が増加します。下の図に示すようにです。


新しい設計では、この制限を回避し、増分 changelog を継続的にアップロードすることでチェックポイント完了時間を高速化できます。詳細な設計を見てみましょう。

2. 設計

Generic Log-Based Incremental Checkpointing のコアアイデアは、State Changelog（状態変更ログ）を導入することです。これにより、より細粒度で状態を永続化できます。

オペレーターは状態を更新する際に 2 つのコピーを書き込みます。1 つの更新は State Table に書き込まれ、もう 1 つのコピーは State Changelog に増分書き込みされます。

チェックポイントは 2 つの部分で構成されます。第 1 の部分はリモートストレージ上に現在永続化されている State Table であり、第 2 の部分は増分 State Changelog です。

State Table の永続化はチェックポイント処理から独立しており、バックグラウンドスレッドによって定期的に永続化されます。このプロセスをマテリアライゼーションと呼びます。

チェックポイント実行時には、新しく追加された State Changelog が永続化されていることを確認するだけです。

新しい設計では、チェックポイント実行時にアップロードが必要なデータ量が非常に少なくなるため、チェックポイントをより安定させ、より頻繁に実行できるようになります。全体のワークフローは下の図に示す通りです。

Generic Log-Based Incremental Checkpointing は、従来のデータベースシステムの WAL 機構に似ています。

データの増分変更（insert/update/delete）はトランザクションログに書き込まれます。この変更されたログの一部が永続ストレージに同期化されると、トランザクションが完了したと見なせます。このプロセスは、上記の方法のチェックポイント処理に似ています。

同時に、データクエリを容易にするために、データ変更は非同期にデータテーブル（Table）にも永続化されます。トランザクションログの関連部分がデータテーブルにも永続化されると、トランザクションログの関連部分を削除できます。このプロセスは、我々の方法の State Table 永続化プロセスに似ています。

WAL に似たこの機構はチェックポイント完了の速度を効果的に向上させますが、いくつかの追加オーバーヘッドももたらします。

追加のネットワーク I/O と追加の Changelog 永続ストレージオーバーヘッド。
Changelog のキャッシュによる追加のメモリ使用量。
フォールトトレラントリカバリ時に Changelog の追加リプレイが必要で、リカバリ時間が増加する可能性。
以下のベンチマーク比較で、これら 3 つの側面の影響も分析します。特にポイント 3 について、Changelog の追加リプレイによるフォールトトレラントリカバリ時間の増加は、より頻繁なチェックポイントによってある程度補償されます。より頻繁なチェックポイントは、フォールトトレラントリカバリ後にリプレイが必要な処理済みデータがより少ないことを意味するためです。

3. Changelog ストレージ（DSTL）

Generic Log-Based Incremental Checkpointing の非常に重要なコンポーネントは、State Changelog ストレージ部分です。これを Durable Short-term Log（DSTL、短期ログ）と呼びます。DSTL は以下の特性を満たす必要があります。

短期永続化

State Changelog はチェックポイントの一部であるため、永続的にストレージに保存する必要があります。同時に、State Changelog は最新の State Table 永続化から現在のチェックポイントまでの Changelog のみを保存すればよいため、短期間（数分）のデータのみを保存すれば済みます。

書き込み頻度が読み取り頻度を大幅に上回る

Changelog の読み取りはリストアまたはリスケール時のみ必要です。ほとんどの場合、追加操作のみがあり、一度書き込まれたデータは変更できません。

非常に短い書き込みレイテンシ

State Changelog はチェックポイントをより高速（1 秒以内）にするために導入されています。したがって、単一の書き込みリクエストは少なくとも想定されるチェックポイント時間以内に完了する必要があります。

一貫性の確保

State Changelog の複数コピーがある場合、複数コピー間で一貫性の問題が発生します。あるコピーの State Changelog が永続化され JM によって確認されると、リストア時にセマンティック一貫性を確保するためにこのコピーを基準として使用する必要があります。

上記の特性から、Changelog ストレージを DSTL 短期ログと命名した理由もわかります。

3.1 DSTL スキームの選択

DSTL は様々な方法で実装できます。分散ログ（Kafka）、分散ファイルシステム（DFS）、データベースなどです。Flink 1.15 でリリースされた Generic Log-Based Incremental Checkpointing MVP バージョンでは、以下の考慮に基づいて DFS を使用して DSTL を実装することを選択しました。

追加の外部依存がない：現在、Flink チェックポイントは DFS に永続化されているため、DFS で DSTL を実装しても追加の外部コンポーネントは導入されません。
追加の状態管理がない：現在の設計では、DSTL の状態管理は Flink チェックポイント機構と統合されているため、追加の状態管理は必要ありません。
DFS はネイティブに永続化と一貫性を提供：複数コピーの分散ログを実装する場合、これらは考慮が必要な追加コストです。

一方、DFS を使用するには以下の欠点があります。

より高いレイテンシ：DFS は通常、ローカルディスクに書き込む分散ログシステムよりもレイテンシが高くなります。
ネットワーク I/O の制限：ほとんどの DFS プロバイダーは、コスト考慮から単一ユーザーの DFS 書き込みのレートとレートを制限しており、極端な場合にネットワーク過負荷を引き起こす可能性があります。
いくつかの予備実験後、現在のほとんどの DFS 実装（S3、HDFS など）のパフォーマンスは使用ケースの 80% を満たすことができ、さらに多くのデータは以下のベンチマークで提供される予定です。

3.2 DSTL アーキテクチャ

以下の図は、RocksDB を例とした DFS ベースの DSTL アーキテクチャ図です。状態更新は Changelog State Backend を通じて二重書き込みされ、1 つは RocksDB に書き込まれ、もう 1 つは DSTL に書き込まれます。RocksDB は定期的にマテリアライゼーションを実行します。つまり、現在の SST ファイルを DFS にアップロードします。一方、DSTL は状態変更を DFS に継続的に書き込み、チェックポイント処理中にフラッシュを完了します。これにより、チェックポイント完了時間はフラッシュが必要なデータ量にのみ依存します。マテリアライゼーションはチェックポイント処理から完全に独立しており、マテリアライゼーションはチェックポイント処理よりもはるかに遅くてもよいことに注意してください。システムのデフォルト値は 10 分です。

ここでは、さらに議論する価値のあるいくつかの問題があります。

状態クリーンアップの問題

前述のように、新しいアーキテクチャでは、チェックポイントは 2 つの部分で構成されます：1）State Table、2）State Change Log。両方の部分は必要に応じてクリーンアップする必要があります。1）この部分のクリーンアップは Flink の既存のチェックポイント機構を再利用します。2）この部分のクリーンアップは比較的複雑です。特に State Change Log の現在の設計では、スモールファイル問題を回避するために、TM を粒度として使用しています。現在の設計では、State Change Log のクリーンアップを 2 つの部分で実行します。まず、Change Log 自体のデータは State Table がマテリアライズされた後に削除する必要があります。既存の Flink チェックポイントクリーンアップ機構 [4]。

DFS 関連の問題

スモールファイル問題
DFS の 1 つの問題は、各チェックポイントで多くの小さなファイルが作成されることであり、Changleog State Backend はより頻繁なチェックポイントを提供できるため、スモールファイル問題はボトルネックになります。この状況を緩和するために、同じ Task Manager 上の同じジョブのすべての State Change を同じファイルに書き込みます。したがって、同じ Task Manager は同じ State Change Log を共有します。

ロングテールレイテンシ問題

DFS の高いロングテールレイテンシに対処するために、DFS 書き込みリクエストが許可されたタイムアウト（デフォルトで 1 秒）以内に完了できない場合、リトライされます。

4. ベンチマークテスト結果の分析

Generic Log-Based Incremental Checkpointing のチェックポイント速度と安定性の向上は、以下の要因に依存します。

State Change Log の増分部分と完全な状態サイズの比率。増分が小さいほど良い。
中断なく状態増分をアップロードする能力。これは状態アクセスモードに関連します。極端な場合、オペレーターがチェックポイント処理前に Flink State Table のみを更新する場合、Changelog はあまり役割を果たしません。
複数タスクからの changelog をグループ化してバッチアップロードする能力。Changelog の DFS へのバッチ書き込みは、作成が必要なファイル数を減らし、DFS の負荷を軽減することで、安定性を向上させます。
ディスクにフラッシュする前に同じキーの更新を重複排除する基盤となる State Backend の能力。状態変更ログは状態更新を保存するため、最終値ではなく、基盤となる State Backend の能力は State Table の完全な状態サイズに対する Changelog 増分の比率を増加させます。
永続ストレージ DFS への書き込み速度。書き込み速度が速いほど、Changelog によってもたらされる向上はあまり顕著でなくなります。

4.1 ベンチマーク設定

ベンチマーク実験では、以下の設定を使用します。

4.2 ValueState Workload

実験の最初の部分は、主に各更新ごとに異なるキー値を持つロードを対象としています。この種のロードは、上記のポイント 2 と 4 の理由により、Changelog の向上がより顕著です。チェックポイント完了時間が 10 倍短縮（99.9 パーセンタイル）、チェックポイントサイズが 30% 増加、リカバリ時間が 66% 〜 225% 増加。下の表に示す通りです。

チェックポイントサイズセクションを詳しく見てみましょう。

表 2：ValueState Workload ベースの Changelog（オン/オフ）のチェックポイント関連指標の比較

Checkpointed Data Size は、Checkpoint Barrier を受信し、チェックポイント処理が開始された後にアップロードされたデータのサイズを指します。Changelog の場合、データの大部分はチェックポイント処理が開始される前に既にアップロードされているため、この指標が Changelog をオンにしている場合とオフにしている場合で大幅に小さい理由です。
Full Checkpoint Data Size は、以前のチェックポイントと共有されるファイルを含む、チェックポイントを構成するすべてのファイルの合計サイズです。通常のチェックポイントと比較すると、Changelog のフォーマットは圧縮されておらず、十分にコンパクトではないため、より多くのスペースを占有します。

4.3 Window Workload

ここでは Sliding Window が使用されています。以下の表に示すように、Changelog はチェックポイント完了時間を約 3 倍高速化します。ただし、ストレージ増幅は大幅に高い（消費スペースが 45 倍に近い）。

表 3：Window Workload ベースの Changelog（オン/オフ）のチェックポイント関連指標の比較

Full Checkpoint Data のストレージスペース増幅の主な理由は以下の通りです。

Sliding Window オペレーターの場合、各データは複数のスライディングウィンドウに追加されるため、複数の更新が発生します。Changelog の書き込み増幅問題はさらに大きくなります。
前述のように、基盤となる State Backend（RocksDB など）がディスクにフラッシュする前に同じキーの更新と重複排除をより強力に行える場合、スナップショットのサイズは Changelog に対して相対的に小さくなります。Sliding Window オペレーターの極端な場合、障害によりスライディングウィンドウがクリアされます。更新とクリアが同じチェックポイント内で発生する場合、そのウィンドウのデータがスナップショットに含まれない可能性が高いです。これはまた、ウィンドウがクリアできる速度が速いほど、スナップショットのサイズは小さくなる可能性が高いことを意味します。

5. 結論

Flink バージョン 1.15 は、Generic Log-Based Incremental Checkpointing の MVP バージョンを実装しました。DFS ベースのこのバージョンは、秒単位のチェックポイント時間を提供でき、チェックポイントの安定性を大幅に向上させますが、ある程度スペースのコストも増加させます。本質的にはスペースと時間を交換しています。バージョン 1.16 では、Local Recovery とファイルキャッシュを通じてリカバリ時間を高速化するなど、本番環境で使用できるようにさらに改善される予定です。一方、Changelog State Backend インターフェースは汎用的であり、同じインターフェースを使用してより高速なストレージに接続することで、より短いレイテンシを実現できます。Apache Bookkeeper などです。さらに、Changelog の他のアプリケーションを研究しており、例えば Changelog を Sink に適用して汎用的なエンドツーエンド exactly-once などを実現することなどです。