Realtime Compute for Apache Flink:チェックポイントとセーブポイント

• ミニバッチが有効な場合に、ステートの TTL が期限切れになった後、データが更新されないのはなぜですか？

• 次の定期的なチェックポイントの開始時刻の計算方法

• VVR 8.x と VVR 6.x における GeminiStateBackend の違い

• フルチェックポイントと増分チェックポイントのサイズが同じになるのは正常ですか？

• Python デプロイメントでチェックポイントが遅い場合の対処法

• チェックポイント例外のトラブルシューティング方法

• エラー: 新しい V4 ステートエンジンを使用して、チェックポイントから古いステートデータを復元しようとしています

• エラー：java.lang.NegativeArraySizeException

• エラー：org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Too many ongoing snapshots.

• エラー: チェックポイントの許容失敗しきい値を超えました

ミニバッチでのデータ更新の失敗

ステートは、以前の完全な計算結果を保持します。ステートの TTL が期限切れになると、ステートはクリアされ、これらの蓄積された結果は失われます。その結果、ミニバッチの結果に基づいて新しいデータを更新できなくなります。

逆に、ミニバッチが無効になっている場合、ステートの TTL が期限切れになると、期限切れのキーのデータが再計算されて出力されます。これにより、継続的なデータ更新が保証されます。ただし、データ更新の頻度が増加すると、処理遅延などの他の問題が発生する可能性があります。

したがって、特定のビジネス要件に基づいて、ミニバッチと TTL の設定を構成する必要があります。

次のチェックポイント開始時刻の計算

次のチェックポイントの開始時刻は、チェックポイント間隔とチェックポイント間の最小休止時間という 2 つのパラメーターによって決まります。新しいチェックポイントは、次の両方の条件が満たされたときにトリガーされます：

• チェックポイント間隔：あるチェックポイントの開始から次のチェックポイントの開始までの最小時間。これは、<前のチェックポイントの開始時刻、次のチェックポイントの開始時刻> 間の時間です。

• 最小休止時間：あるチェックポイントの終了から次のチェックポイントの開始までの最小時間。これは、<前のチェックポイントの終了時刻、次のチェックポイントの開始時刻> 間の時間です。

チェックポイント間隔が 3 分、最小休止時間が 3 分、タイムアウトが 10 分である次の 2 つのシナリオを考えます。

• シナリオ 1：デプロイメントは正常に実行され、すべてのチェックポイントが成功します。

  最初のチェックポイントは 12:00:00 に開始し、12:00:02 に正常に完了します。2 番目のチェックポイントは 12:03:00 に開始されます。

• シナリオ 2：デプロイメントが異常に実行され、チェックポイントがタイムアウトにより失敗します。

  最初のチェックポイントは 12:00:00 に開始し、12:00:02 に正常に完了します。2 番目のチェックポイントは 12:03:00 に開始しますが、タイムアウトのため 12:13:00 に失敗します。3 番目のチェックポイントは 12:16:00 に開始されます。

チェックポイント間の最小休止時間の設定に関する詳細については、「Tuning Checkpointing」をご参照ください。

VVR 8.x と VVR 6.x における GeminiStateBackend の違い

デフォルトでは、VVR 6.x 向けの Realtime Compute for Apache Flink エンジンは GeminiStateBackend V3 を使用し、VVR 8.x エンジンは V4 を使用します。

同じサイズのフルチェックポイントと増分チェックポイント

フルチェックポイントと増分チェックポイントのサイズが同じであることに気付いた場合は、以下を確認する必要があります：

• 増分チェックポイントが正しく設定され、有効になっていることを確認します。

• この動作は、特定のシナリオでは想定される場合があります。例：

  1. データが取り込まれる前 (たとえば 18:29 より前) は、デプロイメントはデータを処理していません。チェックポイントにはソースの初期ステートのみが含まれるため、実質的にフルチェックポイントになります。

  2. 18:29 に 100 万レコードが取り込まれます。このデータが次のチェックポイント間隔 (たとえば 3 分) 内に完全に処理され、他のデータが到着しない場合、最初の増分チェックポイントにはそれらのレコードから生成されたすべてのステートが含まれます。

  この場合、フルチェックポイントと最初の増分チェックポイントは同じサイズになると予想されます。最初の増分チェックポイントには、その時点からの完全な回復を可能にするためにすべてのデータのステートを含める必要があり、これにより機能的にフルチェックポイントと同等になります。

  増分チェックポイントの利点は、通常、2 回目のチェックポイント以降に明らかになります。安定したデータ入力があり、大きなステートの変更がない場合、後続の増分チェックポイントは小さくなるはずであり、システムがステートの変更のみを正しくスナップショットしていることを示します。サイズが同じままである場合は、システムのステートと動作を調査して、潜在的な問題を特定する必要があります。

Python デプロイメントにおけるチェックポイントの遅延

• 原因

  パフォーマンスの低い Python ユーザー定義関数 (UDF) は、チェックポイントの所要時間を増加させ、デプロイメントのパフォーマンスを低下させる可能性があります。

• ソリューション

  バッファーサイズを削減します。[その他の設定] セクションで、次のパラメーターを設定します。手順については、「カスタムデプロイメントパラメーターを設定する」をご参照ください。

  python.fn-execution.bundle.size: デフォルト値：100000。単位：レコード。
  python.fn-execution.bundle.time: デフォルト値：1000。単位：ミリ秒。

  これらのパラメーターの詳細については、「Flink Python Configuration」をご参照ください。

チェックポイント例外のトラブルシューティング

1. 例外タイプの診断

   [アラーム] または [状態] タブでチェックポイントの履歴を表示して、タイムアウトや書き込み失敗などの例外タイプを特定します。

   [概要] タブを選択し、[チェックポイント] セクションを展開します。テーブルには、ID、ステータス、トリガー時間、所要時間、データサイズなど、各チェックポイントの詳細が表示されます。 [ステータス] 列には、各チェックポイントが成功したか失敗したかが表示されます。

2. 問題の切り分けと解決

   • シナリオ 1：頻繁なチェックポイントのタイムアウト。 デプロイメントにバックプレッシャーがないか確認します。バックプレッシャーの根本原因を分析し、遅いオペレーターを特定して、リソースや構成を調整して問題を解決します。詳細については、「バックプレッシャー問題のトラブルシューティング方法」をご参照ください。

   • シナリオ 2：チェックポイントの書き込み失敗。 次の手順に従って、関連する TaskManager のログを見つけ、根本原因を分析します。

     1. [ログ] タブの [チェックポイント] ページで、[チェックポイント履歴] をクリックします。

        [チェックポイント履歴] ページでは、[ID]、[ステータス]、[Acknowledged]、[トリガー時間]、[エンドツーエンドの所要時間]、[チェックポイントデータサイズ] など、各チェックポイントの詳細を表示できます。

     2. 失敗したチェックポイントの横にあるプラス記号 (+) をクリックして、そのオペレーターの詳細を表示します。

     3. 失敗したオペレーターを展開し、[SubTask ID] をクリックして、対応する TaskManager のログに移動します。

エラー：V4 エンジンでの古いステートデータの復元

• エラーメッセージ

  VVR 6.x から VVR 8.x にアップグレードする際に、次のエラーが発生することがあります：You are using the new V4 state engine to restore old state data from a checkpoint

• 原因

  VVR 6.x と VVR 8.x は異なるバージョンの GeminiStateBackend を使用しており、それらのチェックポイントには互換性がありません。

• ソリューション

  この問題を解決するには、次のいずれかの方法を使用できます：

  • 標準フォーマットでセーブポイントを作成し、このステートからデプロイメントを開始します。詳細については、「手動でセーブポイントを作成する」および「デプロイメントの開始」をご参照ください。

  • ステートなしでデプロイメントを再起動します。

  • (非推奨) レガシバージョンの Gemini を引き続き使用します。パラメーター state.backend.gemini.engine.type: STREAMING を設定し、変更を有効にするためにデプロイメントを再起動する必要があります。パラメーターの設定方法については、「デプロイメントパラメーターの設定方法」をご参照ください。

  • (非推奨) VVR 6.x エンジンを引き続き使用してデプロイメントを開始します。

エラー：java.lang.NegativeArraySizeException

• エラーメッセージ

  リストステートを使用するデプロイメントは、ランタイムで次の例外に遭遇する可能性があります：

  Caused by: java.lang.NegativeArraySizeException
    at com.alibaba.gemini.engine.rm.GUnPooledByteBuffer.newTempBuffer(GUnPooledByteBuffer.java:270)
    at com.alibaba.gemini.engine.page.bmap.BinaryValue.merge(BinaryValue.java:85)
    at com.alibaba.gemini.engine.page.bmap.BinaryValue.merge(BinaryValue.java:75)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.internalGet(PageStoreImpl.java:428)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.get(PageStoreImpl.java:271)
    at com.alibaba.gemini.engine.pagestore.PageStoreImpl.get(PageStoreImpl.java:112)
    at com.alibaba.gemini.engine.table.BinaryKListTable.get(BinaryKListTable.java:118)
    at com.alibaba.gemini.engine.table.BinaryKListTable.get(BinaryKListTable.java:57)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.getOrDefault(GeminiSubKeyedListStateImpl.java:97)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.get(GeminiSubKeyedListStateImpl.java:88)
    at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.get(GeminiSubKeyedListStateImpl.java:47)
    at com.alibaba.flink.statebackend.gemini.context.ContextSubKeyedListState.get(ContextSubKeyedListState.java:60)
    at com.alibaba.flink.statebackend.gemini.context.ContextSubKeyedListState.get(ContextSubKeyedListState.java:44)
    at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.onProcessingTime(WindowOperator.java:533)
    at org.apache.flink.streaming.api.operators.InternalTimerServiceImpl.onProcessingTime(InternalTimerServiceImpl.java:289)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:1435)

• 原因

  リストステート内の単一キーのステートデータが 2 GB を超えました。これは次のように発生する可能性があります：

  1. 通常の操作中、リストステート内の単一キーに追加された値は、マージプロセス (たとえば、ウィンドウオペレーター内) を通じて結合され、ステートデータが継続的に増加します。

  2. ステートデータが特定のサイズに達すると、最初に Out Of Memory (OOM) エラーがトリガーされることがあります。デプロイメントが障害から回復した後、マージプロセスにより、ステートバックエンドが 2 GB を超える一時的なバイト配列を要求し、この例外が発生する可能性があります。

  説明

  RocksDBStateBackend でも同様の問題が発生する可能性があり、ArrayIndexOutOfBoundsException またはセグメンテーション違反がトリガーされることがあります。詳細については、「The EmbeddedRocksDBStateBackend」をご参照ください。

• ソリューション

  • 大きなステートがウィンドウオペレーターによって引き起こされている場合は、ウィンドウサイズを小さくすることを検討してください。

  • 大きなステートがデプロイメントのロジックによって引き起こされている場合は、キーを分割するなど、ロジックの再設計を検討してください。

エラー：FlinkKafkaException: Too many ongoing snapshots

• エラーメッセージ

  org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints

• 原因

  このエラーは Kafka sink を使用しているときに発生し、複数の連続したチェックポイントの失敗が原因です。

• ソリューション

  タイムアウトによる失敗を防ぐには、execution.checkpointing.timeout パラメーターを調整して、チェックポイントのタイムアウトを増やします。パラメーターの設定方法については、「カスタムデプロイメントパラメーターの設定」をご参照ください。

エラー: チェックポイントの許容障害しきい値を超えました

• エラーメッセージ

  org.apache.flink.util.FlinkRuntimeException:Exceeded checkpoint tolerable failure threshold.
    at org.apache.flink.runtime.checkpoint.CheckpointFailureManager.handleJobLevelCheckpointException(CheckpointFailureManager.java:66)

• 原因

  設定された許容可能なチェックポイントの失敗回数が低すぎるため、このしきい値を超えるとデプロイメントがフェールオーバーをトリガーします。このパラメーターが設定されていない場合、デフォルト値は 0 であり、チェックポイントの失敗は許容されないことを意味します。

• ソリューション

  execution.checkpointing.tolerable-failed-checkpoints: num パラメーターを設定して、許容されるチェックポイントの失敗回数を調整します。ここで、num は 0 または正の整数である必要があります。パラメーターの設定方法については、「カスタムデプロイメントパラメーターの設定」をご参照ください。