Flink Runtime Architecture

1. ランタイムの概要：

よく知られているように、Flink は分散データ処理フレームワークです。
ユーザーのビジネスロジックはジョブの形式で Flink クラスターに送信されます。
Flink エンジンとして、Flink Runtime はこれらのジョブを正常に実行し、完了させる責務を担います。
これらのジョブはストリーミングジョブまたはバッチジョブのいずれかです。
ベアメタル上でも Flink クラスター上でも実行できます。
Flink Runtime はすべてのタイプのジョブと、さまざまな条件下で実行されるジョブをサポートする必要があります。


1. ジョブの表現方法：

ジョブを実行するには、まず Flink でのジョブの表現方法を理解する必要があります。


ユーザーは API を通じてジョブを記述します。
たとえば、上図の左側に示す StreamWordInput の例では、単語を一つずつ連続で出力します。
その後の Map 操作で単語を 2 タプルにマッピングします。
次に keyBy を接続して同じ単語の 2 タプルをまとめ、sum でカウントし、最後に print で出力します。


左側のジョブは右側の論理トポロジー (StreamGraph) に対応します。
このトポロジーには source、map、sum、print の 4 つのノードがあります。
これらはデータ処理ロジックであり、オペレーターとも呼ばれます。
ノード間の線はデータの配信方法に対応し、データがどのように下流に配信されるかに影響します。
たとえば、map と sum の間の keyBy は、map が生成したデータのうち同じキーを持つデータを必ず同じ下流に配信することを意味します。


StreamGraph を基に、Flink Runtime はさらに JobGraph を生成します。
JobGraph と StreamGraph の違いは、JobGraph が一部のノードを Operator chain として連結する点にあります。
チェーン化の条件は、2 つのオペレーターの同時実行数が同じであり、データ交換方式が one-to-one であることです。
こうして形成された Operator chain は JobVertex とも呼ばれます。


Operator chain の意義は、不要なデータ交換を削減し、チェーン内のオペレーターを同じ場所で実行できるようにすることです。
ジョブの実際の実行時には、論理グラフはさらに ExecutionGraph に変換されます。
ExecutionGraph は論理グラフの同時実行レベルでのビューです。
上図に示すように、下の ExecutionGraph は上の論理グラフのすべてのオペレーターの同時実行数が 2 である場合の表現です。


上図の map と sum がチェーン化できない理由は何でしょうか。
それは、データが複数の下流オペレーターに関わるため、one-to-one のデータ交換方式ではないからです。
論理グラフの JobVertex は複数の同時実行ノードである ExecutionVertex に対応し、各ノードはそれぞれのタスクに対応します。
これらのタスクは最終的に Worker ノードにエンティティとしてデプロイされ、実際のデータ処理ビジネスロジックを実行します。


2. 分散アーキテクチャ：

Flink は分散データ処理フレームワークとして、分散アーキテクチャを備えています。
主に Client、Master、Worker ノードの 3 つの部分に分かれます。


Master は Flink クラスターのメインコントロールセンターです。
1 つ以上の JobMaster を持つことができ、各 JobMaster は 1 つのジョブに対応します。
これらの JobMaster は Dispatcher によって管理されます。
Master ノードにはリソース管理用の ResourceManager もあります。
ResourceManager はすべての Worker ノードを管理し、全ジョブに同時にサービスを提供します。
さらに、Master ノードには Rest Server があり、各種クライアントからの Rest リクエストに応答します。
クライアントにはウェブクライアントとコマンドラインクライアントがあります。
送信できるリクエストには、ジョブの送信、ジョブステータスの照会、ジョブの停止などがあります。
ジョブは ExecutionGraph を通じて個別のタスクに分割され、これらのタスクは最終的に Worker ノードで実行されます。
Worker は TaskExecutor であり、タスク実行のコンテナです。


ジョブ実行の中核コンポーネントは JobMaster、TaskExecutor、ResourceManager の 3 つです。
JobMaster はジョブの管理を担当します。
TaskExecutor は各種タスクの実行を担当します。
ResourceManager はリソースを管理し、JobMaster のリソース要求に対応します。


2. JobMaster：ジョブコントロールセンター：

JobMaster の主な責務には、ジョブのライフサイクル管理、タスクスケジューリング、エラーリカバリ、ステータス照会、分散状態スナップショットが含まれます。


分散状態スナップショットには Checkpoint と Savepoint があります。
Checkpoint は主にエラーリカバリに使用され、Savepoint は主にジョブのメンテナンス (アップグレードや移行など) に使用されます。
分散スナップショットは CheckpointCoordinator コンポーネントによってトリガーされ、管理されます。


JobMaster の中核コンポーネントは Scheduler です。
ジョブのライフサイクル管理、ジョブステータスの維持、タスクスケジューリング、エラーリカバリは、すべて Scheduler が担当します。


1. ジョブライフサイクル管理：

ジョブのライフサイクルの状態と、ジョブのすべての可能な状態遷移は下図に示されています。


通常のフローでは、ジョブは Created、Running、Finished の 3 つの状態を持ちます。
ジョブは最初に Created 状態にあります。
ジョブのスケジューリングが開始されると Running 状態に移行し、タスクのスケジューリングを開始します。
すべてのタスクが正常に完了すると、ジョブは Finished 状態に移行し、最終結果をレポートして終了します。


ただし、ジョブは実行中に問題が発生する可能性があるため、例外処理用の状態も存在します。
ジョブの実行中にジョブレベルのエラーが発生すると、ジョブ全体が Failing 状態に移行し、すべてのタスクをキャンセルします。
すべてのタスクが最終状態 (Failed、Canceled、Finished を含む) になるまで待機してから、エラーの例外を確認します。
例外が回復不可能な場合は、ジョブ全体が Failed 状態に移行して終了します。
例外が回復可能な場合は、Restarting 状態に移行して再起動を試みます。
再起動回数が上限を超えていなければ、ジョブは Created 状態から再スケジューリングされます。
上限に達した場合は、Failed 状態に移行して終了します。
(注意：Flink 1.10 以降のバージョンでは、エラーが発生した際に回復可能な場合、ジョブは Failing 状態を経由せず直接 Restarting 状態に移行します。
すべてのタスクが正常に戻ると、ジョブは Running 状態に戻ります。
回復不可能な場合は、Failing 状態を経て最終的に Failed 状態に移行して終了します。
)

Canceling 状態と Canceled 状態は、ユーザーが手動でジョブをキャンセルした場合にのみ到達します。
ユーザーが Web UI または Flink コマンドでジョブを手動でキャンセルすると、Flink はまず状態を Canceling に遷移させ、その後すべてのタスクをキャンセルします。
すべてのタスクが最終状態に入ると、ジョブ全体が Canceled 状態に移行して終了します。


Suspended 状態は、高可用性が設定されていて、かつ JobMaster がリーダーシップを失った場合にのみ到達します。
この状態は JobMaster に問題が発生して終了したことを示すだけです。
通常、JobMaster が再度リーダーシップを取得するか、別のスタンバイ Master がリーダーシップを取得した後、リーダーシップを取得したノードで再起動します。


2. タスクスケジューリング：

タスクスケジューリングは JobMaster の中核的な責務の 1 つです。
タスクをスケジューリングする際に最初に直面する問題の 1 つは、タスクをいつスケジューリングするかを決定することです。
タスクスケジューリングのタイミングは SchedulingStrategy によって制御されます。
この戦略はイベント駆動型のコンポーネントです。
リッスンするイベントには、ジョブスケジューリングの開始、タスクの状態変化、タスクの出力データが消費可能になること、失敗したタスクの再起動が必要であることが含まれます。
これらのイベントをリッスンすることで、タスクの起動タイミングをより柔軟に決定できます。


現在、さまざまなスケジューリング戦略があります。
Eager と Lazy from sources です。
EagerSchedulingStrategy は主にストリーミングジョブ向けです。
この戦略は、ジョブのスケジューリング開始時にすべてのタスクを直接起動します。
これによりスケジューリング時間を短縮できます。
Lazy from sources は主にバッチジョブ向けです。
この戦略では、ジョブは最初に Source ノードのみをスケジューリングし、どのノードも入力データが消費可能になってから初めて起動されます。
下図に示すように、source ノードがデータの出力を開始して初めて agg ノードが起動され、agg ノードが終了して初めて sink ノードが起動されます。


バッチジョブとストリーミングジョブでスケジューリングポリシーが異なる理由は何でしょうか。
それは、バッチジョブにはブロッキングシャッフルのデータ交換モードがあるためです。
このモードでは、上流がすべてのデータを完全に生成し終えるまで待つ必要があり、その後でなければ下流はこのデータセットを消費できません。
下流を事前に起動しても、リソースを無駄にするだけです。
Eager 戦略と比較すると、バッチジョブのリソースをある程度節約できます。


現在、Pipelined region based という別のスケジューリング戦略が開発中です。
この戦略は Lazy from source 戦略に似ています。
違いは、前者が Pipelined region の粒度でタスクをスケジューリングする点です。


Pipelined region はパイプライン接続されたタスクの集合です。
Pipelined 接続とは、上流と下流のノードがストリーミング方式でデータを交換することです。
つまり、上流が書き込みながら、下流が同時に読み取って消費します。
Pipelined region スケジューリングのメリットは、Eager スケジューリングのメリットをある程度継承できることです。
スケジューリングにかかる時間を短縮し、上流と下流のタスクを並列実行できます。
同時に、Lazy from sources のメリットも保持しており、不要なリソース浪費を回避できます。
複数のタスクをまとめてスケジューリングすることで、同時に実行する必要があるリソース量を把握でき、これに基づいてより深い最適化を実行できます。


(注意：Flink 1.11 以降、Pipelined region 戦略はデフォルトのスケジューリング戦略となり、ストリーミングジョブとバッチジョブの両方にサービスを提供しています。
)

3. タスクスケジューリングのプロセス：

タスクには多くの異なる状態があります。
最初はタスクは Created 状態にあります。
スケジューリングポリシーがタスクの起動を許可すると、Scheduled 状態に移行し、リソース (Slot) の申請を開始します。
Slot を申請した後、Deploying 状態に移行してタスクの記述を生成し、Worker ノードにデプロイしてからタスクを起動します。
起動が成功すると、Worker ノード上で Running 状態に遷移し、JobMaster に通知して、JobMaster 側でもタスク状態を Running に更新します。


無限フローのジョブの場合、最終状態は Running 状態です。
有限フローのジョブの場合、すべてのデータが処理されるとタスクは Finished 状態に移行し、タスクの完了を示します。
例外が発生した場合、タスクは Failed 状態に移行し、影響を受ける他のタスクはキャンセルされて Canceled 状態になる可能性があります。


4. エラーリカバリ：

タスクにエラーが発生した場合、JobMaster の戦略は、失敗したタスクと影響を受ける可能性のあるタスクを再起動することで、ジョブのデータ処理を再開させることです。
これには 3 つのステップがあります。


第 1 ステップは、関連するタスクを停止することです。
エラーで失敗したタスクと、それらから影響を受ける可能性のあるタスクを含みます。
失敗したタスクは既に Failed 状態になっている可能性があります。
その後、他の影響を受けるタスクはキャンセルされ、最終的に Canceled 状態になります。


第 2 ステップは、タスクを Created 状態にリセットすることです。


第 3 ステップは、スケジューリングポリシーに通知してこれらのタスクを再スケジューリングすることです。


前述の影響を受ける可能性のあるタスクについて、どのようなタスクが影響を受けるかは FailoverStrategy によって決定されます。


現在、Flink のデフォルトの FailoverStrategy は RestartPipelinedRegionFailoverStrategy です。
この戦略を採用すると、タスクが失敗した場合、そのタスクが属する region 全体が再起動されます。
これは前述の Pipelined データ交換に関連しています。
Pipelined データ交換のノード間で、いずれかのノードが失敗した場合、他の関連ノードも一緒に失敗します。
これはデータの不一致を防ぐためです。そのため、単一タスクの失敗による複数回のフェイルオーバーを避けるために、一般的な対応は、最初のタスクが失敗した時点で他のタスクもまとめてキャンセルし、その後まとめて再起動することです。

RestartPipelinedRegion 戦略は、失敗タスクが属する Region だけでなく、その下流の Region も再起動します。理由は、タスクの出力が非決定的であることが多いためです。たとえば、あるレコードが下流の 1 番目の並列インスタンスに配信されて再実行され、2 番目の並列インスタンスに配信された際に、2 つの下流が異なる Region にあると、レコードが失われたり、異なるデータが生成されたりする可能性があります。この状況を避けるため、PipelinedRegionFailoverStrategy を使用すると、失敗したタスクが属する Region とそのすべての下流 Region を再起動します。

さらに、RestartAllFailoverStrategy という戦略もあります。この戦略は、任意のタスクが失敗した際にジョブ内のすべてのタスクを再起動します。通常、この戦略はあまり使用されませんが、特別な場合、たとえばタスクが失敗した際にユーザーがローカルリカバリではなく全体としてのリカバリを望む場合、この戦略を使用できます。

3. TaskExecutor：タスク実行ランナー：

TaskExecutor はタスクの実行を担当するコンポーネントです。タスクを実行するためにさまざまなリソースを保持しています。下図に示すように、ここでは主にメモリリソースについて説明します。

すべてのメモリリソースは個別に設定可能です。TaskManager はこれらを階層的に管理しています。最も外側の層は Process Memory で、これは TaskExecutor JVM 全体の総リソース量に相当します。このメモリには JVM 自体が占有するメモリと Flink が占有するメモリが含まれます。Flink が占有するメモリには、フレームワークが占有するメモリとタスクのメモリが含まれます。

タスクが占有するメモリには、Task Heap Memory が含まれます。これはタスクの Java オブジェクトが占有するメモリです。Task Off-Heap Memory は一般的にネイティブのサードパーティライブラリ用に使用されます。Network Memory は Network Buffer の作成に使用され、タスクの入出力に対応します。Managed Memory は管理された Off-Heap Memory で、一部のコンポーネント (オペレーターや StateBackend など) が使用します。これらのタスクリソースは個別に Slot に分割されます。Slot はタスク実行の論理コンテナです。現在、Slot サイズは TaskExecutor 全体のリソースを Slot 数で直接除算して求められます。

1 つの Slot で 1 つ以上のタスクを実行できますが、制約があります。同じ共有グループ内の異なるタイプのタスクのみが同時に 1 つの Slot 内で実行できます。一般的に、同じ PipelinedRegion 内のタスクは 1 つの共有グループに属し、ストリーミングジョブのすべてのタスクも 1 つの共有グループに属します。異なるタイプとは、異なる JobVertex に属する必要があることを意味します。

上図の右側の例に示すように、これは source、map、sink のジョブです。デプロイ後、3 つの Slot のそれぞれに 3 つのタスク (source、map、sum) があります。ただし、1 つの Slot には 2 つのタスクしかありません。source の並列度は 3 であり、これ以上の並列タスクをデプロイできないためです。

SlotSharing の第 1 のメリットは、データ交換のオーバーヘッドを削減できることです。map と sink の間には one-to-one のデータ交換があり、物理的なデータ交換を行うこれらのノードは実際に共有されています。そのため、データ交換をネットワーク経由ではなくメモリ内で実行でき、コストが低くなります。

第 2 のメリットは、ユーザーがリソースを設定しやすいことです。SlotSharing を通じて、ユーザーは n 個の Slot を設定するだけで、sum ジョブの実行を保証できます。n は最大のオペレーターの並列度です。

第 3 のメリットは、各オペレーターの並列度に大きな差がない場合に負荷分散を改善できることです。これは各 Slot に異なるタイプのオペレーターが含まれるため、負荷の高い一部のオペレーターが同じ TaskExecutor に集中するのを防ぎます。

1. タスク実行モデル：

前述のように、各タスクは 1 つの OperatorChain に対応します。一般的に、各 OperatorChain は独自の入力と出力を持ちます。入力は InputGate、出力は ResultPartition です。これらのタスクは一般的に専用スレッドで実行されます。タスクは InputGate からデータを読み取り、OperatorChain に渡します。OperatorChain はビジネスロジック処理を実行し、最後に出力データを ResultPartition に出力します。

例外は Source タスクです。InputGate からデータを読み取るのではなく、SourceFunction を通じて直接データを生成します。上流の ResultPartition と下流の InputGate は Flink の ShuffleService を通じてデータを交換します。ShuffleService はプラグインです。現在、Flink のデフォルトは NettyShuffleService です。下流の InputGate は Netty を使用して上流の ResultPartition からデータを取得します。ResultPartition は複数の SubPartition で構成され、各 SubPartition は 1 つの下流の並列コンシューマーに対応します。InputGate も複数の InputChannel で構成され、各 InputChannel は 1 つの上流の並列インスタンスに対応します。

4. ResourceManager：リソース管理センター：

ResourceManager は Flink のリソース管理センターです。前述のように、TaskExecutor はさまざまなリソースを保持しています。ResourceManager はこれらの TaskExecutor を管理します。新たに起動した TaskExecutor は ResourceManager に登録する必要があり、登録後にそのリソースをジョブの要求に使用できます。

ResourceManager には SlotManager という重要なコンポーネントがあり、Slot の状態を管理しています。これらの Slot 状態は TaskExecutor と ResourceManager 間のハートビートを通じて更新され、ハートビート情報には TaskExecutor 内のすべての Slot の状態が含まれています。ResourceManager は現在のすべての Slot 状態を把握することで、ジョブのリソース申請に対応できます。JobMaster がタスクをスケジューリングする際、ResourceManager に Slot リクエストを送信します。リクエストを受信した ResourceManager は SlotManager に転送し、SlotManager は利用可能な Slot が要求条件を満たすか確認します。満たす場合は対応する TaskExecutor に Slot 申請を開始します。申請が成功すると、TaskExecutor はこの Slot を JobMaster に提供します。

この往復処理を行う理由は、分散による不整合を避けるためです。前述のように、SlotManager の Slot 状態はハートビートを通じて更新されるため、ある程度の遅延があります。さらに、Slot 申請プロセス全体を通じて Slot の状態が変化する可能性があります。そのため、最終的には Slot の offer とその ACK をすべての申請の最終結果として使用する必要があります。

ResourceManager にはさまざまな実装があります。スタンドアロンモードで使用される ResourceManager は StandaloneResourceManager で、ユーザーが手動で Worker ノードを起動する必要があります。この場合、ユーザーは事前にジョブが必要とする総リソース量を把握しておく必要があります。

さらに、自動的にリソースを要求する ResourceManager もあります。YarnResourceManager、MesosResourceManager、KubernetesResourceManager などです。これらの ResourceManager を採用すると、条件が満たされない場合に、ResourceManager は Slot 要求プロセス中に Worker ノードを自動的に起動できます。

YarnResourceManager を例に取ると、JobMaster があるタスクの Slot を要求し、YarnResourceManager はこの要求を SlotManager に渡します。SlotManager が条件を満たす Slot がないと判断した場合、YarnResourceManager に通知します。YarnResourceManager は外部の YarnResourceManager にコンテナを要求し、コンテナを取得した後、その中で TaskExecutor を起動します。TaskExecutor が起動すると、ResourceManager に登録され、利用可能な Slot 情報を通知します。SlotManager はこの情報を取得すると、現在保留中の SlotRequest を満たそうと試みます。満たせる場合は、JobMaster が TaskExecutor に Slot 要求を開始し、要求が成功すると TaskExecutor は Slot を JobMaster に提供します。これにより、ユーザーは最初にジョブのリソース要件を計算する必要がなく、単一の Slot のサイズを確保するだけでよく、タスク実行に対応できます。