Analysis of Flink 1.15's new functional architecture: efficient and stable general incremental checkpoint

1、概要：

Generic Log Based Incremental Checkpointing のオリジナルアイデアは、フル状態スナップショットと増分チェックポイントメカニズムを分離することです。増分 Changelog を継続的にアップロードすることで、各チェックポイントを安定的かつ迅速に完了させ、チェックポイント間隔を短縮して Flink システムのエンドツーエンド遅延を改善できます。大きく分けて 3 つの改善点があります：

1：より短いエンドツーエンド遅延：特に Transactional Sink に対して効果的です。Transactional Sink はチェックポイント完了時に 2 フェーズコミットが完了するため、チェックポイント間隔の短縮はより頻繁なコミットとより短いエンドツーエンド遅延を実現します。

2：より安定したチェックポイント完了時間：従来、チェックポイント完了時間はチェックポイント時に永続化する必要のある(増分)状態のサイズに大きく依存していました。新設計では、増分データを継続的にアップロードすることでチェックポイントフラッシュ時に必要な永続データを削減し、チェックポイント完了の安定性を確保します。

3：障害復旧時のロールバック対象データの削減：チェックポイント間隔が短いほど、障害復旧後の再処理対象データ量も少なくなります。

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

1：チェックポイントバリアのフローとアライメントの速度

2：状態スナップショットを S3 などの不揮発性の高可用性ストレージに永続化するために必要な時間

Flink のチェックポイントメカニズムにあまり詳しくない読者は [1] を参照してください。

Flink 1.12 で導入された Unaligned Checkpoint [2] と Flink 1.14 で導入された Buffer Deblocking [3] は、主に上記の 1 つ目の問題、特にバックプレッシャーが発生している状況での問題を解決します。一方、それより前に導入された増分チェックポイント [4] は、各チェックポイントで必要な永続ストレージ状態のサイズを削減することで 2 つ目の影響要因を低減させます。ただし、実際には完全には解決できません。既存の増分チェックポイントは RocksDB にベースであり、RocksDB はスペースアンプリフィケーションと読み取りパフォーマンスのために定期的にコンパクションを実行します。コンパクションは新しく比較的大きなファイルを生成し、これがアップロード時間を増加させます。Flink ジョブを実行する各物理ノード(Task)には少なくとも 1 つの RocksDB インスタンスが存在するため、物理ノード数の増加に伴ってチェックポイント遅延の確率も上昇します。その結果、Flink の大規模ジョブでは、チェックポイント完了時にノードによる遅延がほぼ毎回発生します。この様子を次の図に示します。

さらに、既存のチェックポイントメカニズムでは、タスクは少なくとも 1 つのチェックポイントバリアを受信してから初めて状態スナップショットを取得し、高可用性ストレージへの永続化を開始するため、チェックポイント完了時間が増加します。この様子を次の図に示します。

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

2、設計：

Generic Log Based Incremental Checkpointing のコアアイデアは State Changelog を導入することで、よりきめ細かい粒度での状態の永続化を可能にします：

1：オペレーターは状態更新時に 2 重書き込みを行います。1 つは State Table に、もう 1 つは State Changelog に増分として書き込まれます。

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

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

4：チェックポイント時には、新しい State Changelog の永続化のみを確保すればよくなります。

新設計では、チェックポイント時のアップロードデータ量が削減されるため、チェックポイントをより安定的に、かつより頻繁に実行できます。全体のワークフローを次の図に示します：

Generic Log Based Incremental Checkpointing は、伝統的なデータベースシステムの WAL メカニズムに似ています：

1：データの増分変更(挿入 / 更新 / 削除)がトランザクションログに書き込まれます。変更ログが永続ストレージに同期された時点で、トランザクションが完了したとみなすことができます。このプロセスは上記のチェックポイントプロセスに似ています。

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

この WAL に似たメカニズムはチェックポイント完了速度を効果的に向上させますが、以下の追加オーバーヘッドもあります：

1：追加のネットワーク I/O と追加の Changelog 永続ストレージのオーバーヘッド

2：Changelog のキャッシュに伴う追加のメモリ使用量

3：障害復旧時に 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 のみを保存すればいいため、短時間(数分間)の保存で十分です。

読み取り頻度よりもはるかに高い書き込み頻度

Restore または Rescale 時のみ Changelog の読み取りが必要です。通常は追加書き込みのみであり、一度書き込まれたデータは変更できません。

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

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

一致性の確保

State Changelog の複数のコピーが存在する場合、複数コピー間の一致性問題が発生します。あるレプリカの State Changelog が永続化され JM によって確認された場合、そのレプリカをベンチマークとして、復旧時のセマンティック一致性を保証する必要があります。

以上の特性からも、Changelog ストレージを DSTL(Short-term Log)と命名している理由がわかります。

3.1 DSTL スキームの選択

DSTL はさまざまな方法で実装できます。たとえば、分散ログ(Kafka)、分散ファイルシステム(DFS)、さらにはデータベースなどです。Flink 1.15 でリリースされる Generic Log Based Incremental Checkpointing の MVP では、以下の理由から DFS を選択しました：

追加の外部依存が不要：現在、Flink のチェックポイントは既に DFS に永続化されているため、DFS で DSTL を実装しても追加の外部コンポーネントが導入されません。

追加の状態管理が不要：現在の設計における DSTL の状態管理は Flink のチェックポイントメカニズムと統合されており、追加の状態管理は不要です。

DFS は永続化と一致性をネイティブに提供：分散ログの複数コピーを実装する場合、これらは追加で検討すべきコストとなります。

一方、DFS の使用には以下のデメリットもあります：

より高いレイテンシ：DFS は一般に、ローカルディスクに書き込む分散ログシステムよりもレイテンシが高くなります。

ネットワーク I/O の制限：大半の DFS プロバイダーはコスト上の理由から、単一ユーザーの DFS 書き込みに対して帯域幅や速度の制限を設けています。極端なケースではネットワークの過負荷が発生する可能性があります。

予備実験の結果、S3 や HDFS などの現在の主要な DFS 実装のパフォーマンスは 80% のユースケースで十分であると考えています。ベンチマークテストでさらに詳細なデータを提供する予定です。

3.2 DSTL アーキテクチャ

次の図は、RocksDB を使用した DFS ベースの DSTL アーキテクチャを示しています。状態更新は Changelog State Backend を通じて RocksDB と DSTL の両方に書き込まれます。RocksDB は定期的にマテリアライズを実行し、現在の SST ファイルを DFS にアップロードします。DSTL は状態変更を継続的に DFS に書き込み、チェックポイント時にフラッシュを完了させます。これにより、チェックポイント完了時間はフラッシュに必要なデータ量のみに依存するようになります。マテリアライゼーションはチェックポイントとは完全に独立したプロセスであり、マテリアライゼーションの頻度はチェックポイントよりもはるかに低く設定できます。システムのデフォルト値は 10 分間です。

以下に、検討すべきいくつかのポイントを紹介します：

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

前述のとおり、新アーキテクチャではチェックポイントは 2 つの部分で構成されます：1)State Table と 2)State Changelog です。両方の部分は必要に応じてクリーンアップする必要があります。1)の部分は既存の Flink チェックポイントメカニズムを再利用します。2)の部分はより複雑です。特に現在の State Changelog 設計では小規模ファイルの問題を回避するため、粒度は TaskManager です。現在の設計では、State Changelog のクリーンアップを 2 つの部分で行います。まず、Changelog データ自体は State Table がマテリアライズされた後に対応する部分を削除する必要があります。また、チェックポイントとなる Changelog の部分のクリーンアップは、既存の Flink チェックポイントクリーンアップメカニズム [4] に統合されています。

DFS 関連の問題

小規模ファイルの問題

DFS の問題の 1 つは、各チェックポイントが多くの小規模ファイルを作成することです。Changelog State Backend はより頻繁なチェックポイントを可能にするため、小規模ファイルの問題がボトルネックになる可能性があります。この状況を緩和するため、同じ TaskManager 上の同じジョブのすべての状態変更を同じファイルに書き込みます。そのため、同じ TaskManager は同じ State Changelog を共有します。

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

DFS の高いロングテールレイテンシの問題を解決するため、許容タイムアウト(デフォルトは 1 秒)内に完了できない DFS 書き込みリクエストはリトライされます。

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

Generic Log Based Incremental Checkpointing によるチェックポイントの速度と安定性の改善は、以下の要因に依存します：

状態変更ログの増分とフル状態サイズの比率。増分が小さいほど良くなります。

状態増分を継続的にアップロードする能力。これは状態アクセスモードに関係します。極端なケースでは、オペレーターがチェックポイント直前にのみ Flink の State Table を更新する場合、Changelog は顫著な効果を発揮しません。

複数タスクからの Changelog をグループ単位でバッチアップロードする能力。Changelog のグループバッチ書き込みは DFS へのファイル作成数を削減し、DFS 負荷を低減させることで安定性を向上させます。

ディスクフラッシュ前に同じキーへの更新を重複排除する基盤の State Backend の能力。State Changelog は最終値ではなく状態更新を保存するため、基盤の State Backend の能力は Changelog 増分の State Table に対する比率を大きくします。

永続ストレージ DFS への書き込み速度。書き込みが速いほど、Changelog による改善効果は相対的に小さくなります。

4.1 ベンチマーク設定

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

オペレーターパラレリズム：50

実行時間：21 時間

State Backend：RocksDB(増分チェックポイント有効)

永続ストレージ：S3(Presto プラグイン)

マシンモデル：AWS m5.xlarge(TM あたり 4 スロット)

チェックポイント間隔：10 ミリ秒

State Table マテリアライゼーション間隔：3 分

入力レート：50K イベント / 秒

4.2 ValueState ワークロード

実験の最初の部分では、毎回異なるキー値を持つワークロードにフォーカスします。上記の 2 つ目および 4 つ目の理由により、Changelog の改善効果は顮著です。チェックポイント完了時間は 10 倍短縮(99.9 パーセンタイル)、チェックポイントサイズは 30% 増加、復旧時間は 66% から 225% 増加しました。詳細は次の表に示します。

表 1：ValueState ワークロードにおける Changelog ベースの指標比較

チェックポイントサイズをさらに詳しく見ていきましょう：

表 2：ValueState ワークロードにおける Changelog オープン / クローズのチェックポイント関連指標比較

Checkpointed Data Size は、チェックポイントバリアを受信してチェックポイントプロセスを開始した後のアップロードデータのサイズを指します。Changelog の場合、チェックポイントプロセス開始前に既に大半のデータがアップロード済みです。これが、Changelog 有効時のこの指標がクローズ時よりもはるかに小さい理由です。

Full Checkpoint Data Size は、前のチェックポイントと共有されるファイルを含む、チェックポイントを構成するすべてのファイルの合計サイズです。通常のチェックポイントと比較して、Changelog 形式は圧縮されておらずコンパクトではないため、より多くのスペースを消費します。

4.3 ウィンドウワークロード

ここではスライディングウィンドウを使用しています。次の表に示すように、Changelog はチェックポイント完了時間を約 3 倍高速化します。ただし、ストレージのアンプリフィケーションははるかに高く(約 45 倍のスペースを消費)なります：

Full Checkpoint Data のストレージアンプリフィケーションの主な原因は以下のとおりです：

スライディングウィンドウオペレーターでは、各データが複数のスライディングウィンドウに追加されるため、複数回の更新が発生します。Changelog のライトアンプリフィケーションの問題はさらに大きくなります。

前述のとおり、基盤の State Backend(RocksDB など)がディスクフラッシュ前に同じキーの更新を重複排除する能力が強いほど、スナップショットのサイズは Changelog に対して相対的に小さくなります。スライディングウィンドウオペレーターの極端なケースでは、障害発生時にスライディングウィンドウがクリアされます。更新とクリアが同じチェックポイント内で発生した場合、このウィンドウのデータがスナップショットに含まれない可能性があります。これは、ウィンドウのクリアが速いほど、スナップショットサイズも小さくなる可能性があることを意味しています。

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 を実現することを検討しています。