Flink's new generation of stream computing and fault tolerance

1. 高可用性ストリームコンピューティングの重要パス

上図の両方向の軸は、時間の経過に伴うビッグデータアプリケーションの遅延のマップです。右に行くほど時間遅延要件が短くなり、左に行くほど遅延要件が低くなります。Flink が誕生した当初は、おそらく上図の中央付近に位置していました。ストリームコンピューティングは右側に対応し、バッチコンピューティングは左側に対応すると理解できます。過去 1〜2 年で、Flink のアプリケーショングラフは左側に向けて大きく拡張されました。これがよく言われるストリーム・バッチ統合です。同時に、よりリアルタイムな方向への推進も止まっていません。

Flink はストリームコンピューティングから始まりましたが、よりリアルタイムな方向へ前進するとはどういう意味でしょうか。よりリアルタイムで極端なストリームコンピューティングとは何でしょうか。

通常処理の場合、Flink エンジンフレームワーク自体は、定期的にチェックポイントスナップショットを取得する以外はほとんど追加オーバーヘッドがなく、チェックポイントスナップショットの大部分は非同期であるため、通常処理での Flink は非常に効率的で、エンドツーエンド遅延は約 100 ミリ秒です。効率的な処理をサポートする必要があるため、Flink はフォールトトレラント復旧とリスケーリング時に比較的高いコストがかかります。ジョブ全体を停止し、過去のスナップショットチェックポイントから全体として復元する必要があります。このプロセスには約数秒かかります。ジョブ状態が比較的大きい場合、分レベルに達します。ウォームアップや他のサービスプロセスの開始が必要な場合、時間はさらに長くなります。

したがって、Flink の極端なストリームコンピューティングの重要なポイントはフォールトトレラント復旧部分です。ここで言及する究極のストリームコンピューティングは、レイテンシ、安定性、一貫性にある程度の要件があるシナリオを指します。たとえば、リスク管理やセキュリティなどです。これは Fault Tolerance 2.0 が解決を目指す問題でもあります。

2. Fault Tolerance 2.0 と重要な問題

フォールトトレラント復旧はフルリンクの問題であり、障害検出、ジョブキャンセル、新しいリソース申請スケジューリング、状態復旧と再構築などを含みます。同時に、既存の状態から復旧したい場合は、通常処理中にチェックポイントを実行し、通常処理に影響を与えないほど軽量にする必要があります。

フォールトトレランスは多次元の問題でもあります。異なるユーザーと異なるシナリオでは、フォールトトレランスに対する要件が異なり、主に以下の側面を含みます。

データ整合性（Data Consistency）、オンライン機械学習などの一部のアプリケーションは部分的なデータ損失を許容できます。

レイテンシ、一部のシナリオではエンドツーエンドレイテンシに対する要件がそれほど高くないため、通常処理とフォールトトレラント復旧時に行う作業を平均化できます。

復旧動作（Recovery Behavior）、たとえば大画面やレポートのリアルタイム更新は、完全に迅速に復旧する必要はなく、むしろ最初のデータを迅速に復旧することが重要です。

コスト（Cost）、ユーザーは自身のニーズに応じてフォールトトレランスに対して異なる価格を支払う意思があります。要約すると、異なる角度からこの問題を考慮する必要があります。

さらに、フォールトトレランスは Flink エンジン側だけの問題ではありません。Flink とクラウドネイティブの組み合わせは Flink の将来の重要な方向であり、クラウドネイティブへの依存がフォールトトレランスの設計と方向を決定づけています。非常にシンプルな弱依存を通じてクラウドネイティブがもたらす利点を活用することを期待しています。たとえば、リージョンを跨いだ耐久性など、最終的にはネイティブなステートレスアプリケーションのようにステートフル Flink アプリケーションを弾力的にデプロイできるようにしたいと考えています。

上記の考慮に基づいて、Flink Fault Tolerance 2.0 には異なる重点と方向があります。

まず、スケジューリングの観点から、エラー復旧のたびに、グローバルスナップショットに対応するすべてのタスクノードがロールバックされるのではなく、障害が発生した単一または一部のノードのみが復旧されます。これは、ウォームアップが必要またはシングルノードの初期化に時間がかかるシナリオ、たとえばオンライン機械学習シナリオにおいて 必要です。これに関連する一部の作業、たとえば Approximate Task-local Recovery は VVP で開始されており、Exactly-once Task-local Recovery についてもいくつかの成果を上げています。

次に、チェックポイントとクラウドネイティブに関連する部分に焦点を当てます。

3. Flink におけるデータ復旧プロセス

では、フォールトトレランスは実際に何を解決するのでしょうか。私の意見では、その本質はデータ復旧の問題を解決することです。

Flink データは大きく分けて以下の 3 つのカテゴリに分類できます。1 つ目はメタ情報で、Flink ジョブの実行に必要な最小情報セットであり、チェックポイントアドレス、Job Manager、Dispatcher、Resource Manager などを含みます。フォールトトレランスは Kubernetes/ZooKeeper などのシステムの高可用性によって保証され、議論しているフォールトトレランスの範囲外です。Flink ジョブが実行されると、データソースからデータを読み取り Sink に書き込みます。中間を流れるデータは、処理された中間データ Inflight Data（2 番目のカテゴリ）と呼ばれます。集約オペレーターなどのステートフルオペレーターの場合、入力データを処理した後、オペレーター状態データ（3 番目のカテゴリ）が生成されます。

Flink は定期的にすべてのオペレーターの状態データのスナップショットを取り、耐久性のある安定した大量ストレージ（Durable Bulk Store）にアップロードします。このプロセスをチェックポイントと呼びます。Flink ジョブでエラーが発生すると、以前のチェックポイントスナップショットにロールバックしてチェックポイント復旧を行います。

現在、チェックポイント処理の効率を向上させるための多くの作業を行っています。実際の作業では、エンジン層でのほとんどの Oncall またはワークオーダー問題は基本的にチェックポイントに関連しており、さまざまな理由でチェックポイント処理がタイムアウトするためです。

チェックポイント処理のプロセスを簡単に振り返りましょう。この部分に詳しい方は直接スキップできます。チェックポイント処理のプロセスは以下のステップに分けられます。

ステップ 1: チェックポイントコーディネーターが Source 側からチェックポイントバリアを挿入します（上図の黄色い縦棒）。

ステップ 2: バリアは中間データ処理とともに下流に流れます。オペレーターを通過する際、システムはオペレーターの現在の状態の同期スナップショットを取得し、スナップショットデータを非同期でリモートストレージにアップロードします。これにより、バリアより前のすべての入力データがオペレーターの状態に反映されます。オペレーター状態が大きい場合、チェックポイント処理の完了に影響します。

ステップ 3: オペレーターに複数の入力がある場合、オペレーターはスナップショットの取得を開始する前にすべての入力バリアを取得する必要があります。これが上図の青いボックスの部分です。アライメントプロセス中にバックプレッシャーがある場合、中間処理のデータフローが遅くなり、バックプレッシャーのないラインもブロックされ、チェックポイントが非常に遅く、あるいは不可能になることがあります。

ステップ 4: すべてのオペレーターの中間状態データがリモートの安定したストレージへのアップロードに成功した後、完全なチェックポイントが真に完了します。

これら 4 つのステップから、チェックポイント処理を迅速かつ安定して行うことに影響を与える 2 つの主要な要因があることがわかります。1 つは中間データ処理の流れが遅いことで、もう 1 つはオペレーター状態データが大きすぎてアップロードが遅いことです。以下、これら 2 つの要因を解決する方法について説明します。

4. 安定した高速効率的なチェックポイント処理

中間データの流れが遅い場合、以下の方法があります。

中間データにブロックされない方法を見つける: Unaligned Checkpoint - Source から挿入された Barrier のスキップされた中間データを直接 Sink にプッシュします。
または中間データを十分に少なくする: Buffer Debloating。
状態データが大きい場合、チェックポイントを行うたびにアップロードされるデータ状態を十分に小さくする必要があります: Generalized Log-Based Incremental Checkpoint。
以下、各ソリューションについて詳しく説明します。

4.1 Unaligned Checkpoint

Unaligned Checkpoint の原理は、Source から挿入された Barrier のスキップされた中間データを即座に Sink にプッシュし、スキップされたデータをスナップショットに一緒に格納することです。したがって、Unaligned Checkpoint の場合、その状態データにはオペレーターの状態データだけでなく、処理された中間データも含まれ、上図の黄色いボックスのように、Flink パイプライン全体の完全な瞬間的なスナップショットとして理解できます。Unaligned Checkpoint は非常に迅速にチェックポイントを行えますが、追加のパイプラインチャネル中間データを格納する必要があるため、格納する必要のある状態は大きくなります。Unaligned Checkpoint は昨年 Flink-1.11 バージョンでリリースされました。Flink-1.12 と 1.13 バージョンでは、Unaligned Checkpoint のリスケーリングと、Aligned Checkpoint から Unaligned Checkpoint への動的切り替えがサポートされています。

4.2 Buffer Debloating

Buffer Debloating の原理は、スループットとレイテンシに影響を与えずに、アップストリームとダウンストリームのキャッシュ内のデータを削減することです。観察の結果、オペレーターは大きな入出力バッファーを必要としないことがわかりました。データフローが遅いときにジョブがパイプライン全体を満たす場合を除き、ジョブメモリが不足して OOM が発生する以外、データを過度にキャッシュしてもほとんど役に立たず。

ここで簡単な見積もりができます。各タスクについて、出力であれ入力であれ、私たちが持つバッファーの総数は、大まかに各チャネルに対応する排他バッファーの数にチャネル数を掛けたものに、共通のフローティングバッファーの数を加えたものです。バッファーの総数に各バッファーのサイズを掛けると、結果はローカルバッファープールの合計サイズになります。そしてシステムデフォルト値を代入して計算すると、並列度が少し大きく、データシャッフルが数回多いと、ジョブ全体の中間を流れるデータが簡単に数ギガバイトに達することがわかります。

実際には、それほど多くのデータをキャッシュする必要はありません。オペレーターがアイドルにならないことを保証するのに十分なデータだけが必要です。これがまさに Buffer Debloating の役割です。Buffer Debloating は、アップストリームとダウンストリームのバッファーの合計サイズを動的に調整し、パフォーマンスに影響を与えずにジョブに必要なバッファーサイズを最小限に抑えます。現在の戦略は、アップストリームがダウンストリームで約 1 秒で処理できるデータを動的にキャッシュすることです。さらに、Buffer Debloating は Unaligned Checkpoint にも効果的です。Buffer Debloating は中間を流れるデータを削減するため、Unaligned Checkpoint がスナップショットを取得する際、追加格納が必要な中間データも削減されます。

上図は、バックプレッシャー下での Buffer Debloating の場合の、Debloat Target に応じたチェックポイント処理時間の変化の比較図です。Debloat Target は、アップストリームキャッシュの「期待時間」内にダウンストリームで処理できるデータを指します。この実験では、Flink ジョブには合計 5 つの Network Exchange があるため、合計チェックポイント処理時間は約 Debloat Target の 5 倍に等しくなり、実験結果とほぼ一致しています。

4.3 Generalized Log-Based Incremental Checkpoint

前述のように、状態のサイズもチェックポイント処理の完了時間に影響します。Flink のチェックポイント処理は、同期スナップショットと非同期アップロードの 2 つの部分で構成されているためです。同期プロセスは通常非常に高速で、メモリ内の状態データをディスクにフラッシュするだけで十分です。しかし、状態データを非同期にアップロードする部分はアップロードデータ量に関連しているため、各スナップショットでアップロードする必要のあるデータ量を制御するために Generalized Log-Based Incremental Checkpoint を導入しました。

ステートフルオペレーターの場合、内部状態が変更されると、更新は State Table に記録されます。上図のようにです。チェックポイントが発生すると、RocksDB を例にとると、State Table がディスクにフラッシュされ、ディスクファイルが非同期でリモートストレージにアップロードされます。チェックポイントモードに応じて、アップロードされる部分はフルチェックポイントまたはチェックポイントの増分部分になります。しかし、どのモードであっても、アップロードファイルのサイズは State Backend ストレージに強く縛られます。たとえば、RocksDB は増分チェックポイントもサポートしていますが、多層コンパクションがトリガーされると、多くの新しいファイルが生成されます。この場合、増分部分は完全なチェックポイントよりも大きくなり、アップロード時間は依然として制御できません。

アップロードプロセスがチェックポイント処理のタイムアウトを引き起こしているため、アップロードプロセスをチェックポイント処理プロセスから分離することで問題を解決できます。これがまさに Generalized Log-Based Incremental Checkpoint が行いたいことです。本質的に、チェックポイント処理プロセスと State Backend ストレージコンパクションを完全に分離します。

具体的な実装方法は以下の通りです。ステートフルオペレーターの場合、State Table に状態更新を記録するだけでなく、State Changelog にも増分を書き込み、非同期でリモートストレージにフラッシュします。これにより、チェックポイントは 2 つの部分で構成されるようになります。第 1 の部分はリモートストレージに現在マテリアライズされて格納されている State Table で、第 2 の部分はまだマテリアライズされていない増分部分です。したがって、実際にチェックポイントを行う際、アップロードする必要のあるデータ量は少なくなり、より安定します。チェックポイントをより安定させるだけでなく、より頻繁に行うこともできます。エンドツーエンド遅延を大幅に短縮できます。特に Exactly Once Sink の場合、2 フェーズコミットを完了する前に完全なチェックポイントを完了する必要があるためです。

5. クラウドネイティブフォールトトレランスと弾力的スケーリング

クラウドネイティブの文脈では、迅速な拡張と縮小は Flink にとって大きな課題です。特に Flink-1.13 バージョンで Re-active Scaling モードが導入されて以来、Flink ジョブは頻繁にスケーリングイン/アウトを行う必要があるため、リスケーリングは Re-active スケーリングモードの主要なボトルネックになっています。リスケーリングとフェイルオーバーが解決すべき問題は大部分で類似しています。たとえば、マシンが削除された後、システムは迅速に認識し、再スケジューリングして状態を復旧する必要があります。もちろん、違いもあります。フェイルオーバー時には、状態を復旧してオペレーターに状態を引き戻すだけで済みます。しかし、リスケーリング時には、トポロジーが並列度を変更するため、状態を再配布する必要があります。

状態を復旧する際、まずリモートストレージから状態データをローカルに読み取り、次に読み取ったデータに基づいて状態を再配布する必要があります。上図のように、状態がやや大きい場合、このプロセス全体は単一并列度で 30 分以上かかります。そして実際には、状態再割り当てに必要な時間は、リモートストレージから状態データを読み取る時間よりもはるかに長いことがわかりました。

では、状態はどのように再配布されるのでしょうか。Flink の状態は Key Group を最小単位としてセグメント化されており、状態の Key Space を 0 から始まる正の整数のセットにマッピングするものとして理解できます。この正の整数のセットが Key Group Range です。この Key Group Range はオペレーターが許可する最大並列度に関連しています。上図のように、オペレーターの並列度を 3 から 4 に変更すると、再割り当てされた Task1 の状態は元の 2 つの Task 状態の一部から連結され、この連結された状態は連続しており交差がありません。そのため、この特徴を利用して最適化を行うことができます。

上図からわかるように、最適化後、DB Rebuild 部分の最適化効果はまだ非常に明白ですが、この部分の作業はまだ探索段階にあり、未解決の問題が多く残っているため、当面は明確なコミュニティプランはありません。

最後に、この記事の内容を簡単に振り返りましょう。まず、フォールトトレランスがなぜ必要なのかを議論しました。フォールトトレランスは Flink ストリームコンピューティングの重要パスだからです。次に、フォールトトレランスに影響を与える要因を分析しました。フォールトトレランスはフルリンクの問題であり、障害検出、ジョブキャンセル、新しいリソース申請スケジューリング、状態復旧と再構築などを含み、多次元からこの問題について検討し多角的に考慮する必要があります。現在の焦点は主に、どのように安定して迅速にチェックポイントを行うかです。多くの実際の問題がチェックポイントに関連しているためです。最後に、クラウドネイティブの文脈でフォールトトレランスと弾力的スケーリングを統合する方法について、いくつかの探索的作業を議論しました。