Optimization and Practice of Flink Unaligned Checkpoint in Shope

1. チェックポイントの問題

1.1 チェックポイントの技術的な問題

Flink の運用におけるバックプレッシャーがチェックポイントのタイムアウト失敗を深刻に引き起こすことは、Flink 本番環境でよく見られる問題です。継続的なバックプレッシャーは、チェックポイントが長期間失敗し続ける原因となります。

たとえば、外部クエリや書き込みのパフォーマンスボトルネック、CPU ボトルネック、ラッシュ時やピーク時のデータスキューなどの一般的なシナリオは、間接的にチェックポイントの失敗を引き起こします。

1.2 チェックポイントの継続的な失敗がビジネスに与える影響

・30 分間のラグデータを消費した後、開発者がその 30 分間のタスクの消費レートが遅く、期待に沿わないことを発見しました。タスクの並列度を上げて再起動し、消費能力を向上させたいと考えましたが、チェックポイントが失敗し続ける場合、30 分前のチェックポイントから回復する必要があります。この 30 分間に消費されたデータは繰り返し消費され、リソースの浪費やビジネスデータの重複リスクが生じます。

・ラグ消費時に、許容できる失敗チェックポイント数（デフォルトの CP 失敗許容回数は 0）が低すぎると、Flink ジョブが無限ループに陥る可能性があります（ラグ消費によりジョブに深刻なバックプレッシャーが発生 → 深刻なバックプレッシャーによりチェックポイントがタイムアウト失敗 → チェックポイント失敗によりジョブが失敗 → ジョブ失敗によりさらにラグ消費が増加）。この結果、ラグを永遠に消費できません。

・チェックポイント失敗の無制限な許容は適切な解決策ではありません。許容回数が高すぎる場合：

⚠ 本番環境の問題をタイムリーに発見できません。

⚠ 一部のコネクターはチェックポイント時にデータやファイルをコミットします。チェックポイントが失敗し続けると、これらのデータやファイルが長期間コミットされず、データの遅延やトランザクションタイムアウトを引き起こします。たとえば、Kafka Producer のトランザクションタイムアウトはトランザクション失敗につながります。

⚠ ジョブを再起動すると、大量のデータが繰り返し消費されます。

・ビジネスのピーク時やプロモーション時もラグ消費と同様で、同じ問題が発生します。

1.3 非整列チェックポイントの導入

上記の背景から、多くのユーザーは Flink タスクにボトルネック（深刻なバックプレッシャー）がある場合でもチェックポイントを成功させたいと考えています。そこで Flink コミュニティは FLIP-76 で非整列チェックポイント機構（以下 UC）を導入しました。

2. 非整列チェックポイントの原理

2.1 UC のコアアイデア

バックプレッシャーが深刻な場合、整列チェックポイント（以下 AC）のタイムアウトは、主にバリアがデータストリーム内でキューイングされることが原因です。バックプレッシャーが深刻な場合、データフローが非常に遅くなり、バリアのフローも遅くなるため、最終的に AC タイムアウトに至ります。

UC のコアアイデアは、データフローが非常に遅い場合、バリアが何らかの機構を通じてデータを追い越し、Source から Sink まで一気に 빠르게到達できるようにすることです。

2.2 タスク UC プロセスの詳細

現在のタスクの上流タスクの並列度が 3、下流タスクの並列度が 2 であると想定します。UC 開始後、このタスクの 3 つの InputChannel は上流から送信されたバリアを次々に受信します。

図に示すように、グレーのボックスはバッファー内のデータを表しています。InputChannel-0 が最初にバリアを受信し、他の InputChannel はまだバリアを受信していません。

InputChannel がバリアを受信すると、UC の最初のフェーズ、つまり UC 同期フェーズを直接開始します。注意点：

・バリアがタスクのネットワーク層の入力バッファーに入った時点で、タスクは直接 UC を開始します。

・他の InputChannel がバリアを受信するのを待たず、その InputChannel 内のバリアより前のデータを処理する必要もありません。

下の図に示すように、データ整合性を確保するため、UC 同期フェーズではタスクはデータを処理できません。同期フェーズでは以下の処理を行います。

・バリア追い越し：バリアをすべての ResultSubPartition の先頭に送信し、すべての入出力バッファーを追い越して、バリアを下流タスクに素早く送信できます。

・バッファーのスナップショット：追い越したすべての入出力バッファーのスナップショットを取得します。

・オペレーターの snapshot State メソッドを呼び出します。

・Flink エンジンがオペレーター内部の状態のスナップショットを取得します。

いくつかの留意点があります。

・UC 中、バリアが追い越したバッファーデータは直接スキップされます。データが失われないよう、これらのバッファーは状態と一緒に HDFS に書き込む必要があります。チェックポイントから回復する際、これらのデータは消費されます。

・同期フェーズではタスクはデータを処理できません。ブロッキング時間を最小化するため、同期フェーズではバッファーと状態データへの参照のみを行います。データの HDFS への実際の書き込みは非同期で完了されます。

・UC 同期フェーズの最後の 2 ステップは AC と完全に一致し、オペレーター内部の状態をスナップします。

UC 同期フェーズの完了後、タスクはデータ処理を再開し、同時に UC の第 2 フェーズであるバリア整列および UC 非同期フェーズを開始します。非同期フェーズでは、同期フェーズのシャローコピーされた状態とバッファーを HDFS に書き込みます。

なぜ UC にバリア整列が必要なのか？

タスクが UC を開始した時点では、多くの InputChannel がバリアを受信しておらず、これらの InputChannel のバリアより前にはスナップショットが必要なネットワークバッファーが存在する可能性があります。そのため、UC の第 2 フェーズでは、すべての InputChannel のバリアが到着する必要があり、バリアより前のバッファーのスナップショットを取得する必要があります。UC が HDFS に書き込むデータは 3 種類あると考えられます。

・同期フェーズで参照されたすべての入出力バッファー。

・同期フェーズで参照されたオペレーター内部の状態。

・同期後に他の InputChannel のバリアより前にあるバッファー。

非同期フェーズでこれら 3 種類のデータがすべて書き込まれた後、ファイルアドレスを JobManager に報告し、現在のタスクの UC が終了します。

注：理論上、UC 非同期フェーズのバリア整列は高速です。前述のタスクで示したように、バリアはすべての入出力バッファーを素早く追い越して下流タスクにバリアを送信できます。したがって、上流タスクも同様です。バリアはすべての上流バッファーを追い越して現在のタスクに素早く送信されます。

2.3 UC 実践における問題

バリアがタスクのネットワーク層の入力バッファーに入った時点で、タスクは直接 UC を開始します。バリアはすべてのバッファーを素早く追い越して下流に送信されるため、UC はバックプレッシャーの影響を受けません。理論上：バックプレッシャーがどれだけ深刻であっても、UC バリアは一気に追い越し、Source から Sink まで素早く流れ、各タスクはスナップショットを素早く完了できます。

理論は非常に優れていますが、実際の研究とタスク使用の過程で、UC の効果は期待通りではないことがわかりました。

多くのシナリオで、タスクのバックプレッシャーが深刻な場合、UC は依然として成功せず、UC の期待されるメリットが大幅に低下しています。

UC は HDFS に書き込まれるファイル数を大幅に増加させ、オンラインサービスの安定性に影響を与え、大規模適用の難易度を高めます。

UC にはいくつかのバグがあります。

以降のセクションでは、上記の問題、Shopee のソリューション、およびコミュニティへの貢献について紹介します。

3. UC のスループットを大幅に向上

タスクはデータ処理中にチェックポイントを処理できません。現在処理中のデータを処理し、結果を OutputBufferPool に書き込んでから、InputChannel が UC バリアを受信したかどうかを確認する必要があります。受信していた場合、UC が開始されます。

タスクが 1 つのデータを処理して結果を OutputBufferPool に書き込むのに 10 分以上かかる場合、UC は依然としてタイムアウトします。通常、1 つのデータの処理自体は遅くありませんが、結果の OutputBufferPool への書き込みに時間がかかる場合があります。

OutputBufferPool の観点から見ると、上流タスクがプロデューサーで下流タスクがコンシューマーです。したがって、下流タスクにボトルネックがある場合、上流タスクが OutputBufferPool に結果を出力する際にバッファー待ちでスタックし、UC を開始できません。

この問題を解決するため、Flink コミュニティは FLINK-14396 で予約バッファーの機構を導入しました。このソリューションでは、タスクがデータを処理する前に OutputBufferPool にアイドルバッファーがあるかどうかを確認します。アイドルバッファーがない場合は待ち続けます。詳細なプロセスは下の図に示すとおりです。

OutputBufferPool に空きバッファーができるまでデータを処理しないことで、タスクがデータを処理した後、結果を OutputBufferPool にスムーズに書き込め、ステップ 5 のデータ出力フェーズでスタックしなくなります。最適化後にアイドルバッファーがない場合、タスクはステップ 3 のアイドルバッファーと UC バリアの待機フェーズでスタックします。このフェーズでは UC バリアを受信すると、UC を素早く開始できます。

3.1 1 つのデータ処理に複数のバッファーが必要なシナリオの改善

下の図に示すように、予約されるバッファーは 1 つだけなので、単一データの処理に複数のバッファーが必要な場合、タスクはデータ処理中にステップ 5 で結果を OutputBufferPool に出力する際に依然としてスタックし、UC を処理できない可能性があります。

たとえば、大規模な単一データ、flatmap、ウィンドウトリガー、ブロードキャストウォーターマークなど、すべて 1 つのデータ処理に複数のバッファーが必要なシナリオです。これらのシナリオでは、タスクはステップ 5 のデータ出力フェーズでスタックし、UC のパフォーマンスが低下します。この問題を解決するコアアイデアは、タスクをステップ 5 ではなくステップ 3 の待機フェーズでスタックさせる方法です。

上記の問題に基づき、Shopee は FLIP-227 でオーバードラフトバッファを提案しました。アイデアは次のとおりです。データ処理中にバッファーが不足し、TaskManager に空きネットワークメモリがある場合、現在のタスクの OutputBufferPool が TM からバッファーをオーバードラフトして、データ処理のステップ 5 を完了します。

注：OutputBufferPool はアイドルバッファーがない場合にのみオーバードラフトバッファを使用できます。したがって、オーバードラフトバッファが使用されると、次のラウンドのステップ 3 でバリアとアイドルバッファーを待つフェーズに入った際、タスクは OutputBufferPool にアイドルバッファーがないと判断します。すべてのオーバードラフトバッファが下流タスクに消費され、OutputBufferPool に少なくとも 1 つのアイドルバッファーができるまで、タスクはデータ処理を再開しません。

デフォルトでは、taskmanager.network.memory.max-overdraft-buffers-per-gate=5 です。つまり、タスクの各 OutputBufferPool は TM から 5 つのバッファーをオーバードラフトできます。オーバードラフトバッファ機構の導入後、TM のネットワークメモリが十分な場合、1 つのデータ処理に 5 つのバッファーが必要でも、UC はまったくスタックしません。TM のネットワークメモリが大きい場合、パラメーターを増やしてより多くのシナリオに対応できます。

FLINK-1.16 からオーバードラフトバッファ機能がサポートされました。関連する JIRA には FLINK-27522、FLINK-26762、FLINK-27789 があります。

3.2 レガシーソースの改善

データソースの観点から、タスクは SourceTask と非 SourceTask の 2 種類に分かれます。

SourceTask は外部コンポーネントから Flink ジョブにデータを読み取り、非 SourceTask は InputChannel からデータを読み取ります。データは上流タスクから来ます。

非 SourceTask は InputChannel からデータを読み取る前に OutputBufferPool を確認し、空きバッファーがある場合にのみ読み取ります。SourceTask が外部コンポーネントからデータを読み取る前に OutputBufferPool にアイドルバッファーがあるかどうかを確認しない場合、UC のパフォーマンスは低下します。

Flink には 2 種類のソース、レガシーソースと新ソースがあります。

新ソースとタスクの動作モードはプルモードです。つまり、タスクがソースからデータをプルします。動作モードは InputChannel と似ており、タスクはソースからデータをプルする前に OutputBufferPool に空きバッファーがあることを確認します。

レガシーソースはプッシュモードです。つまり、レガシーソースは外部コンポーネントからデータを読み取り、直接下流に送信します。OutputBufferPool にアイドルバッファーがない場合、レガシーソースはスタックし、UC を正常に処理できません。

しかし、本番環境のほぼすべての Flink ジョブは依然としてレガシーソースを使用しています。レガシーソースは Flink コミュニティで廃止され、もはやメンテナンスされていないため、Shopee はよく使われるレガシーソースに改善を加えました。

改善のアイデアは上記と同様です。レガシーソースは OutputBufferPool に空きバッファーがあることを確認してから下流にデータを送信します。

Flink で最もよく使われる FlinkKafkaConsumer は実際にはレガシーソースであり、業界の多くの Flink ユーザーが依然としてレガシーソースを使用しています。内部で改善したレガシーソースのバージョンを FLINK-26759 としてコミュニティに共有しました。

4. UC のリスクを大幅に低減

上記の最適化により、バックプレッシャーが深刻な場合でも、レガシーソースで 1 つのデータ消費に複数のバッファーが必要なシナリオでも UC が素早く成功するようになり、一部の Flink ユーザーの期待に沿う効果に達しました。ただし、UC は依然として大規模本番運用の基準を満たしていません。主な理由は、UC は AC と比較してネットワークバッファーをチェックポイントに書き込むため、追加のリスクが導入されるからです。

より多くのファイルが HDFS に書き込まれ、NameNode に追加の負荷がかかります。

データのスキーマがアップグレードされた後、シリアル化に互換性がない場合、データを回復できません。

オペレーター間の接続が変更されると、オペレーター間のバッファーデータを回復できません（たとえば、rebalance から forward に変更する場合）。

4.1 AC から UC へのスムーズな切り替え

ユーザーはこれらのリスクを回避しつつ、UC のメリットを享受したいと考えています。そこで Flink コミュニティは整列チェックポイントタイムアウト機構を導入しました。つまり、デフォルトのチェックポイントは AC で、指定時間内に AC が完了しない場合、UC に切り替えます。

AC タイムアウト機構の導入後、UC のリスクは完全には回避されていませんが、タスクにバックプレッシャーがない場合は依然として AC で、追加のリスクはありません。深刻なバックプレッシャーで AC が失敗した場合に UC に切り替えることで、チェックポイントの成功を確保します。

AC タイムアウト=1 分、チェックポイントタイムアウト=5 分と想定します。つまり、チェックポイントは依然として AC で開始し、AC が 1 分間失敗した場合に UC に切り替え、チェックポイントの合計持続時間が 5 分を超えるとタイムアウト失敗となります。

AC タイムアウトの開発には 3 つの段階があります。最初の 2 段階は目標を達成できませんでした。つまり、1 分経過してもジョブは AC から UC に切り替えられず、5 分経っても UC に切り替えられず、チェックポイントがタイムアウト失敗する結果となりました。これらの 3 段階を目標とともに理解できます。

4.2 InputChannel の AC から UC への切り替えサポート

FLINK-19680 は AC タイムアウト機構を初めてサポートしました。第 1 段階の原理は、各タスクが最初のバリアを受信した時点から計時を開始し、タスク内の AC バリア整列時間が AC タイムアウトを超えた場合、現在のタスクが AC から UC に切り替えます。

この機構の問題は、ジョブ内のタスクが多い場合です。Source から Sink まで 10 個のタスクがあるとします。10 個のタスクの内部バリア整列時間がそれぞれ 59 秒の場合、どのタスクも UC に切り替えられませんが、10 個すべてのタスクが整列する必要があるため、合計チェックポイント時間は少なくとも 590 秒（5 分以上）必要となり、チェックポイントは依然としてタイムアウト失敗します。

第 1 段階の問題に基づき、FLINK-23041 は改善を加えました。第 2 段階の原理は、バリアにチェックポイント開始時のタイムスタンプを搭載します。InputChannel がバリアを受信した後、現在のシステム時刻からチェックポイント開始時の時刻を引いて、チェックポイントがどのくらい経過したかを示します。

1 分以上経過している場合、直接 UC に切り替えます。

1 分未満の場合、1 分から AC に消費された時間を引いて、UC に切り替えるまでの残り時間を示します。タイマーを設定し、時間になったら UC に切り替えます。

第 1 段階と比較して、第 2 段階は複数のタスクの時間蓄積の問題を解決しました。InputChannel がバリアを受信し、指定時間内に AC が完了しなければ、AC から UC に定期的の切り替えができます。

4.3 出力バッファーの AC から UC への切り替えサポート

第 2 段階の完了後、InputChannel は AC から UC への切り替えを十分にサポートしていると考えられます。しかし、問題も明らかです。出力バッファーが AC から UC への切り替えをサポートしていません。

タスクのバックプレッシャーが深刻な場合、バリアは出力バッファーでキューイングされます。バリアが 5 分以内に下流タスクの InputChannel に送信できない場合、チェックポイントは依然としてタイムアウトします。

この問題に基づき、Shopee は FLINK-27251 と FLINK-28077 で出力バッファーの AC から UC への切り替えをサポートする改善を提案しました。設計思想は次のとおりです。

UC が有効で現在 AC の場合、バリアは出力バッファーの末尾に送信されます。しかし、しばらくした後、AC から UC に変換する必要が生じる可能性があるため、タイマーを設定する必要があります。

タイマーの時間が経過した時点でバリアが依然として出力バッファーでキューイングされている場合、AC は UC に変換されます。バリアが出力バッファーの先頭に追い越し、図中の水色のバッファーはスナップショットによりチェックポイントに書き込む必要があります。

コミュニティは前期にチェックポイントのパフォーマンスを評価するベンチマークを設計しました。下の図に示すとおりです。最適化が Flink master のブランチにマージされた後、UC のパフォーマンスは 11 倍に向上しました。

4.4 UC の小ファイル統合

AC タイムアウト機構を有効にした後、Flink はバックプレッシャーが深刻でない場合は AC を使用し、バックプレッシャーが深刻な場合はスムーズに UC に切り替えることができます。これにより、UC の追加リスクが大幅に低減され、バックプレッシャーが深刻な場合でも UC のメリットを享受できます。しかし、大規模本番運用では依然としてリスクが存在します。

デフォルトでは、Flink の各サブタスクはバッファー用に 1 つのファイルを書き込みます。10 個のタスクがあり、各タスクの並列度が 1000 と想定すると、UC は 10,000 個の追加小ファイルを書き込む可能性があります。Kafka クラスターが故障したりボトルネックが生じたりすると、大量の Flink ジョブの Kafka への書き込みが遅くなり、多数の Flink タスクが AC から UC に切り替わります。この場合、大量のタスクが瞬時に数十万の小ファイルを HDFS に書き込み、NameNode のアバランシェを引き起こす可能性があります。

小ファイルの問題を解決するため、Shopee は FLINK-26803 と FLINK-28474 で UC の小ファイルをマージする改善を提案しました。

最適化のアイデア：複数のタスクが同じファイルを共有します。各タスクは個別にファイルを作成するのではなく、CheckpointStreamManager からファイルストリームを取得します。

CheckpointStreamManager は N 個のタスクに 1 つのファイルを割り当てます。デフォルトでは、channel-state.number-of-tasks-share-file=5 です。つまり、5 個のタスクが 1 つの UC ファイルを共有する場合、UC ファイルの数は 5 分の 1 に削減されます。複数のタスクが同時に同じファイルに書き込むとスレッドセーフの問題が発生するため、ファイル書き込み時にファイルストリームをロックして、複数のタスクが直列にファイルを書き込むようにします。

本番環境の経験から、大量の UC 小ファイルは 1 MB 以下であるため、20 個のタスクで 1 つのファイルを共有することは許容範囲です。もちろん、NameNode の負荷が非常に低く、Flink ジョブが書き込み効率をより重視する場合は、このパラメーターを 1 に設定できます。これはタスクが UC ファイルを共有しないことを意味します。

現在、UC の小ファイル統合機能をコミュニティに貢献し続けています。

4.5 ネットワークバッファーのデッドロック修正

UC に対する Shopee の貢献には、FLINK-22946 でのネットワークバッファーのリサイクル時のデッドロック問題の解決も含まれます。

5. UC の Shopee における本番実践と将来計画

5.1 UC の本番実践

UC がもたらす追加リスクを回避するため、Shopee 内部では整列チェックポイントタイムアウトを 1 分に設定しています。これは、タスクのバックプレッシャーが深刻でない場合、AC が 1 分以内に完了すれば AC を使用することを意味します。バックプレッシャーが深刻で AC が 1 分以内に完了しない場合は UC に切り替えます。

Shopee Flink プラットフォームの開発ページには UC スイッチも追加されています。ユーザーはジョブに対して非整列チェックポイントを有効にするかどうかを選択できます。現在までに数百の Flink タスクが UC を有効にしており、UC を使用している現在の操作は良好に動作しています。UC はバックプレッシャー下でも成功します。

5.2 UC の将来計画

ユーザーが UC で遭遇する問題を引き続き注視していきます。数か月の安定運用後、AC タイムアウトを有効にする前提で、すべてのタスクに対して UC を有効にすることを検討できます。

Shopee の内部バージョンでは、Flink のスケジューリングとネットワークメモリモジュールを大幅に変更しました。TM が必要とするネットワークメモリを正確に計算できます。将来的には、UC のオーバーヘッドバッファー用に個別のメモリを予約する予定です。