Automatic derivation of parallelism for Flink batch jobs

I. はじめに

多くのユーザーにとって、Flink オペレータの適切な並列度を設定することは容易ではありません。バッチジョブの場合、並列度が低すぎると実行時間が長くなり、障害からの回復も遅くなります。一方、必要以上に大きな並列度を設定するとリソースの無駄遣いにつながり、タスクのデプロイメントやデータシャッフルのオーバーヘッドも増加します。

バッチジョブの実行時間を制御するためには、オペレータの並列度は処理すべきデータ量に比例しているべきです。ユーザーは、処理すべきデータ量を見積もって並列度を設定する必要があります。しかし、処理すべきデータ量を正確に見積もるのは非常に困難です。処理すべきデータ量は毎日変化する可能性がありますし、ジョブ内に多数の UDF や複雑なオペレータが存在する場合、出力データ量を判断するのが困難になります。

この問題を解決するため、Flink 1.15 で新しいスケジューラを導入しました。Adaptive Batch Scheduler です。このアダプティブバッチジョブスケジューラは、各オペレータが処理すべき実際のデータ量に基づいて、ジョブの実行中に自動的に並列度を導出します。これにより、以下のメリットがもたらされます。

* バッチジョブの同時実行調整の煩雑さを大幅に軽減
* 処理データ量に応じて異なるオペレータに異なる並列度を設定でき、以前はグローバルな並列度しか設定できなかった SQL ジョブにとって特に有益
* 日ごとに変化するデータ量により良く適応できる

2. 使用方法

Flink でオペレータの並列度を自動導出するには、以下の設定が必要です。

アダプティブバッチジョブスケジューラを有効化する
オペレータの並列度を -1 に設定する

2.1 アダプティブバッチジョブスケジューラの有効化

アダプティブバッチジョブスケジューラを有効化するには、以下の設定が必要です。

jobmanager.scheduler: AdaptiveBatch を設定する
execution.batch-shuffle-mode を ALL-EXCHANGES-BLOCKING（デフォルト）に設定する。現在、アダプティブバッチジョブスケジューラはシャッフルモードが ALL-EXCHANGES-BLOCKING のジョブのみサポートしているためです。

さらに、自動導出されるオペレータ並列度の上限と下限、各オペレータの想定処理データ量、ソースオペレータのデフォルト並列度を指定する関連設定もあります。詳細は Flink ドキュメント [1] を参照してください。

2.2 オペレータの並列度を -1 に設定する

アダプティブバッチジョブスケジューラは、ユーザーが並列度を明示的に指定していないオペレータ（つまり、並列度がデフォルト値 -1 のオペレータ）の並列度のみを推定します。そのため、以下の設定が必要です。

* parallelism.default: -1 を設定する
* SQL ジョブの場合、table.exec.resource.default-parallelism: -1 を設定する必要がある
* DataStream/DataSet ジョブの場合、ジョブ内でオペレータの setParallelism() メソッドを通じて並列度を指定しないこと
* DataStream/DataSet ジョブの場合、StreamExecutionEnvironment/ExecutionEnvironment の setParallelism() メソッドを通じてジョブ内で並列度を指定しないこと

3. 実装の詳細

次に、アダプティブバッチジョブスケジューラの実装の詳細について説明します。その前に、関連するいくつかの用語概念を簡単に紹介します。

論理ノード（JobVertex）[2] と論理トポロジー（JobGraph）[3]：論理ノードは、パフォーマンス向上のために複数のオペレータを連結して形成されたオペレータチェーンです。論理トポロジーは、複数の論理ノードがデータフロー図として接続されたものです。

実行ノード（ExecutionVertex）[4] と実行トポロジー（ExecutionGraph）[5]：実行ノードはデプロイ可能な物理タスクに対応し、論理ノードが並列度に従って展開されることで生成されます。たとえば、論理ノードの並列度が 100 の場合、100 個の対応する実行ノードが生成されます。実行トポロジーは、すべての実行ノードの接続によって構成される物理実行グラフです。

上記の概念の詳細については、Flink ドキュメント [6] を参照してください。なお、アダプティブバッチジョブスケジューラは、論理ノードの並列度を導出することで、そのノードに含まれるオペレータの並列度を決定します。

実装の詳細は主に以下の部分から成ります。

* スケジューラが実行ノードの出力データサイズを収集できるようにする
* 処理すべきデータ量に応じて論理ノードの並列度を計算する新しいコンポーネント VertexParallelismDecider [7] を導入する
* 実行トポロジーの動的構築をサポートする。つまり、実行トポロジーは空の実行トポロジーから始まり、ジョブのスケジューリングに伴って徐々に実行ノードを追加していく
* アダプティブバッチジョブスケジューラを導入して、実行トポロジーの更新とスケジューリングを行う

3.1 実行ノードの出力データ量の収集

アダプティブバッチジョブスケジューラは、論理ノードが処理すべきデータ量に基づいてその並列度を決定するため、上流ノードが生成したデータ量を収集する必要があります。このため、numBytesProduced カウンターを導入して、各実行ノードが生成したデータパーティション（ResultPartition）のデータ量を記録し、実行ノードの実行完了時にその累積値をスケジューラに送信します。

3.2 論理ノードの適切な並列度の決定

論理ノードの並列度を計算する新しいコンポーネント VertexParallelismDecider を導入しました。計算アルゴリズムは以下の通りです。

前提条件

* V はユーザーが設定した各実行ノードの想定処理データ量
* totalBytesnon-broadcast は論理ノードが処理すべき非ブロードキャストデータの総量
* totalBytesbroadcast は論理ノードが処理すべきブロードキャストデータの総量
* maxBroadcastRatio は各実行ノードが処理するブロードキャストデータの上限比率
* normalize(x) は x に最も近い 2 のべき乗を出力する関数

注目すべきは、この数式に 2 つの特別な処理を導入している点です。

* 各実行ノードが処理するブロードキャストデータの割合を制限する
* 並列度を 2 のべき乗に調整する

さらに、上記の数式はソースノードの並列度を直接決定するには使用できません。ソースノードはデータを消費しないためです。この問題を解決するため、jobmanager.adaptive-batch-scheduler.default-source-parallelism という設定オプションを導入しました。これにより、ユーザーはソースノードの並列度を手動で設定できます。なお、このオプションが必須でないソースも存在します。一部のソースは自身で並列度を推定できるためです（たとえば HiveTableSource。詳細は HiveParallelismInference を参照）。これらのソースについては、自身で並列度を推定する方が推奨されます。

3.2.1 各実行ノードが処理するブロードキャストデータの割合の制限

数式内で、各実行ノードが処理するブロードキャストデータの上限比率を maxBroadcastRatio に制限しています。つまり、各実行ノードは少なくとも (1-maxBroadcastRatio) V の非ブロードキャストデータを処理します。もしこれを行わないと、ブロードキャストデータ量が V に近い場合、非ブロードキャストデータ量が非常に少なくても、大きな並列度が計算される可能性があります。これは不要であり、リソースの無駄やタスクデプロイメントコストの増加を招きます。

通常、実行ノードが処理すべきブロードキャストデータ量は、処理すべき非ブロードキャストデータ量よりも少なくなります。そのため、デフォルトで maxBroadcastRatio を 0.5 に設定しています。現在、この値はコード内にハードコードされていますが、将来的には設定可能にすることを検討します。

3.2.2 並列度を 2 のべき乗に調整する

normalize 関数は、データスキューを導入しないように、並列度を最も近い 2 のべき乗に調整します。この部分をより良く理解するために、まずサブパーティションの動的マッピングセクションを読むことをお勧めします。

図 4(b) を例にとると、A1/A2 は 4 つのサブパーティションを生成し、B は最終的に並列度 3 に決定されます。この場合、B1 は 1 つのサブパーティションを消費し、B2 は 1 つのサブパーティションを消費し、B3 は 2 つのサブパーティションを消費します。異なるサブパーティションのデータ量は同じであると仮定すると、B3 が消費すべきデータ量は B1/B2 の 2 倍となり、データスキューが発生します。

この問題を解決するため、すべての下流実行ノードが消費するサブパーティション数を同じにする必要があります。つまり、上流が生成するサブパーティション数は、下流論理ノードの並列度の整数倍であるべきです。簡便のため、ユーザーが指定する最大並列度を 2^N とします（もしそうでない場合は、自動的に設定値を超えない 2^N に調整されます）。そして、下流論理ノードの並列度を最も近い 2^M（M <= N）に調整します。これにより、サブパーティションが下流で均等に消費されることが保証されます。

ただし、これは暫定的な解決策であり、最終的には自動負荷分散を通じて解決されるべきです。これは後続のバージョンで実装する予定です。

3.3 実行トポロジーの動的構築

アダプティブバッチジョブスケジューラが導入される前は、実行トポロジーは静的な方法で構築されていました。つまり、スケジューリング開始前に実行トポロジーが完全に作成されていました。論理ノードの並列度をランタイムで決定できるようにするため、実行トポロジーは動的構築をサポートする必要があります。

3.3.1 実行トポロジーへのノードとエッジの動的追加

実行トポロジーの動的構築とは、Flink ジョブが空の実行トポロジーから始まり、スケジューリングに伴って徐々に実行ノードを追加していくことを意味します。図 2 に示す通りです。

実行トポロジーは実行ノードと実行エッジ（ExecutionEdge）から構成されます。論理ノードが展開されて実行ノードが作成され、実行トポロジーに追加されるのは、以下の条件が満たされる場合のみです。

論理ノードに対応する並列度が既に決定されている（これにより Flink は何個の実行ノードを作成すべきかを把握できる）
すべての上流論理ノードが展開済みである（これにより Flink は新規作成された実行ノードを上流実行ノードと実行エッジで接続できる）

3.3.2 サブパーティションの動的マッピング

アダプティブバッチジョブスケジューラが導入される前は、実行ノードをデプロイする際、Flink は下流論理ノードの並列度を把握する必要がありました。なぜなら、下流論理ノードの並列度は、上流実行ノードが生成すべきサブパーティション数を決定するからです。図 3 を例にとると、下流 B の並列度は 2 です。そのため、上流 A1/A2 は 2 つのサブパーティションを生成する必要があります。インデックス 0 のサブパーティションは B1 によって消費され、インデックス 1 のサブパーティションは B2 によって消費されます。

しかし明らかに、これは動的グラフには適していません。上流実行ノードがデプロイされる時点では、下流論理ノードの並列度がまだ決定されていない可能性があるためです（つまり、A1/A2 がデプロイされる時点では、B の並列度はまだ決定されていません）。この問題を解決するため、上流実行ノードが生成するサブパーティション数と下流論理ノードの並列度のデカップリングが必要です。

以下の方法でデカップリングを実現します。上流実行ノードが生成するサブパーティション数を、下流論理ノードの最大並列度（最大並列度は設定可能な固定値）に設定します。そして、下流論理ノードの並列度が決定された後、これらのサブパーティションを異なる下流実行ノードに均等に分配して消費させます。つまり、下流実行ノードをデプロイする際、各下流実行ノードには消費用のサブパーティション範囲が割り当てられます。N を下流論理ノードの並列度、P をサブパーティション数とすると、k 番目の下流実行ノードが消費するサブパーティションの範囲は以下のようになります。

3.4 実行トポロジーの動的更新とスケジューリング

アダプティブバッチジョブスケジューラのジョブスケジューリング方法は、基本的にデフォルトのスケジューラと同じです。唯一の違いは、アダプティブバッチジョブスケジューラが空の実行トポロジーからスケジューリングを開始する点です。スケジューリングイベントを処理する前に、すべての論理ノードの並列度を決定し、その後、論理ノードに対応する実行ノードを生成し、実行エッジを通じて上流ノードと接続して実行トポロジーを更新しようとします。

スケジューラは、各スケジューリング前にトポロジー順序に従ってすべての論理ノードの並列度を決定しようとします。

ソースノードの場合、その並列度はスケジューリング前に決定されます
非ソースノードの場合、すべての上流ノードのデータ出力が完了するまで並列度を決定できません
その後、スケジューラはトポロジー順序に従って論理ノードを展開し、実行ノードを生成しようとします。展開可能な論理ノードは、以下の条件を満たす必要があります。

論理ノードの並列度が既に決定されている
すべての上流論理ノードが展開済みである
4. 将来の展望 - 自動負荷分散

バッチジョブの実行時、データスキューが発生する可能性があります（特定の実行ノードが他の実行ノードよりもはるかに多くのデータを処理する必要がある）。これにより、ジョブのロングテール現象が発生し、ジョブの完了速度が低下します。Flink がこの問題を自動的に改善または解決できれば、ユーザーにとって大きな助けになります。

データスキューの典型的なケースは、一部のサブパーティションのデータ量が他のサブパーティションよりも著しく大きい場合です。この問題は、より細かい粒度のサブパーティションに分割し、サブパーティションのサイズに応じてワークロードを分散することで解決できます（図 5）。アダプティブバッチスケジューラの作業は、この方向への第一歩と考えることができます。自動リバランスの要件はアダプティブバッチスケジューラの要件と類似しており、両方とも動的グラフのサポートと ResultPartition サイズの取得を必要とするためです。

アダプティブバッチジョブスケジューラの実装に基づき、最大並列度を増加させること（より細かい粒度のサブパーティションのため）と、サブパーティション範囲分割アルゴリズムを単純に変更すること（ワークロード分散のため）で、上記の問題を解決できます。現在の設計では、サブパーティションの範囲はサブパーティション数に応じて分割されています。これを、サブパーティション内のデータ量に応じて分割するように変更できます。これにより、各サブパーティションのデータ量をおおむね均等にでき、下流実行ノードのワークロードを分散できます。