Realtime Compute for Apache Flink:Flink バッチ処理チューニングガイド

実行モード

• ストリームジョブ: ストリーム処理モードは、継続的で無限のデータストリームの処理に重点を置いています。その主な目的は、低レイテンシーのデータ処理を実現することです。このモードでは、データはノード間ですぐに渡され、パイプラインで処理されます。したがって、ストリームジョブのすべてのノードのサブタスクは、同時にデプロイおよび実行されます。

• バッチジョブ: バッチ処理モードは、有界データセットの処理に重点を置いています。その主な目標は、高スループットのデータ処理を提供することです。この実行モードでは、ジョブは通常、複数のステージで構成されます。独立したステージは並列で実行でき、リソース使用率を向上させます。データ依存関係のあるステージの場合、ダウンストリームタはアップストリームタスクが完了するのを待ってから開始する必要があります。

データ転送

• ストリームジョブ: 低レイテンシーを実現するために、ストリームジョブは中間データをメモリに保持し、永続化せずにネットワーク経由で直接転送します。ダウンストリームノードの処理能力が不十分な場合、アップストリームノードにバックプレッシャーを引き起こす可能性があります。

• バッチジョブ: バッチジョブは、ダウンストリームタスクで使用するために、中間結果を外部ストレージシステムに書き込みます。デフォルトでは、これらの結果ファイルは TaskManager のローカルディスクに保存されます。リモートシャッフルサービスが使用されている場合、データファイルはリモートシャッフルサービスに保存されます。

リソース要件

• ストリームジョブ: ストリームジョブは、起動時にすべてのリソースを事前に割り当てる必要があります。これにより、すべてのサブタスクを同時にデプロイして実行できます。

• バッチジョブ: バッチジョブは、実行時にすべてのリソースを一度に取得する必要はありません。Flink は、入力データの準備が整うとタスクをバッチでスケジュールできます。これにより、既存のリソースをより効率的に使用できます。ジョブは限られたリソース (単一のスロットを含む) でもスムーズに実行できます。

タスクの失敗と再起動

• ストリームジョブ: 障害が発生した場合、ストリームジョブは最新のチェックポイントまたはセーブポイントから回復できます。これにより、ジョブの進行状況のロールバックが最小限に抑えられます。ただし、中間結果は永続化されないため、回復中にすべてのタスクを再起動する必要があります。

• バッチジョブ: バッチジョブは中間結果をディスクに保存します。タスクが失敗して再起動した場合、これらの中間結果を再利用できます。つまり、失敗したタスクとそのダウンストリームタスクのみを再起動する必要があり、グローバルなロールバックは必要ありません。これにより、障害後に再実行する必要があるタスクの数が減り、回復効率が向上します。ただし、バッチジョブにはチェックポイントメカニズムがないため、これらの再起動されたタスクは最初から実行する必要があります。

リソース構成

並列度の構成

ジョブのリソース構成で、グローバルな並列度を設定したり、自動並列度推論を有効にしたりできます。

• グローバル並列度: グローバル並列度は、ジョブ内で並列に実行できるタスクの最大数を決定します。ページのジョブの並列度を直接入力できます。ジョブはこの値をグローバルなデフォルトの並列度として使用します。

• 自動推論: 自動推論が有効になっている場合、Flink バッチジョブは自動的に並列度を推論します。これは、各ノードによって消費される合計データ量を分析し、各サブタスクによって処理されると予想される平均データ量を分析することによって行われます。これにより、並列度構成を最適化できます。

さらに、Realtime Compute Engine Ververica Runtime (VVR) 8.0 以降では、次の設定項目が提供されています。ジョブの実行時パラメーター設定エリアでこれらを構成して、自動並列度推を微調整できます:

バッチジョブの TaskManager の数が並列度を超えることがある

バッチジョブでは、TaskManager (TM) はオンデマンドで動的に作成および解放されます。例:

• ジョブの並列度は 16 で、各 TM には 1 スロットのみが割り当てられます。

• 最初のオペレーターが開始すると、16 個の TM が作成されます。

• 一部のサブタスクが早期に終了した場合、対応する TM は一定期間非アクティブになった後に自動的に解放されます。

• 後続のオペレーターが実行を開始すると、システムは新しい TM を要求します。これにより、作成された TM の総数 (17、18、19 など) が元の並列度を超えます。

これは例外ではありません。これはバッチジョブのエラスティックなスケジューリングの通常の動作です。

TM の総数を厳密に制限するには、 [最大スロット数] パラメーターを構成できます。

並列度とスロット数の違い

並列度: ジョブ内の各オペレーターで同時に実行できるサブタスクの数を定義します。これは処理能力の理論上の上限です。
スロット: Flink のリソースユニット。1 つのスロットで 1 つのサブタスクを実行できます。

• ストリームジョブ: デフォルトでは、スロット共有が有効になっています。起動時に、すべてのタスクがすぐに実行されるように、グローバル並列度と同じ数のスロットを要求します。

• バッチジョブ: すべてのリソースを事前に割り当てる必要はありません。実際の並列タスク数は、グローバル並列度が高くても、現在利用可能なスロットによって制限されます。

例: 並列度が 4 のストリームジョブには 4 つのスロットが必要です。クラスターに利用可能なスロットが 4 つしかない場合、バッチジョブは最大 4 つのサブタスクを実行できます。残りのタスクは、スロットが解放されるのを待ってから実行されます。

バッチジョブがスタックする原因を特定する方法

TaskManager のメモリ、CPU、およびスレッドの使用状況を監視する方法については、「ジョブのパフォーマンスを表示する」をご参照ください。

• メモリ問題のトラブルシューティング: まず、メモリ使用量を確認して、メモリ不足による頻繁なガベージコレクション (GC) が発生しているかどうかを判断します。メモリが不足していることを確認した場合は、TaskManager のメモリ構成を増やして、頻繁な GC によるパフォーマンスの問題を軽減します。

• CPU 使用率の分析: 個々のスレッドが大量の CPU リソースを消費していないか確認します。これがジョブのスタッタリングの原因である可能性があります。

• スレッドスタックのトレース: スレッドスタック情報を使用して、現在のノードの実行におけるボトルネックを分析します。

エラー: デバイスに空き容量がありません

リアルタイムコンピューティングエンジンでバッチジョブを実行するときに「No space left on device」エラーが発生した場合、通常、TaskManager が中間結果ファイルを保存するために使用するローカルディスク領域が使い果たされたことを示します。各 TaskManager で使用可能なディスク領域は限られており、割り当てられた CPU コアに比例します。

解決策：

• 各 TaskManager のスロット数を減らします。これにより、単一ノードでの並列タスク数が減り、ローカルディスク領域への要求が減少します。

• TaskManager の CPU コア数を増やします。これにより、TaskManager のディスク領域が増加します。