Flink Execution Engine

1. 背景

インターネットとモバイルインターネットの継続的な発展に伴い、あらゆる業界で膨大なビジネスデータが蓄積されてきました。
ユーザー体験の向上と市場でのプロダクト競争力強化のため、企業はリアルタイム方式でビッグデータを処理するようになっています。
ソーシャルメディアのリアルタイム大画面、E コマースのリアルタイムレコメンデーション、City Brain のリアルタイム交通予測、金融業界のリアルタイム不正防止など、これらのプロダクトの成功は、ビッグデータのリアルタイム処理が止められないトレンドになっていることを示しています。


リアルタイム化の全体的なトレンドの中で、Flink はリアルタイムコンピューティング業界のデファクトスタンダードとなりました。
Alibaba だけでなく、国内外の各分野のリーディングメーカーも Flink をリアルタイムコンピューティングの技術基盤として使用していることがわかります。


リアルタイムビジネスは出発点に過ぎません。
Flink の目標の 1 つは、ユーザーにリアルタイムとオフラインの統合ユーザー体験を提供することです。
実際、多くのユーザーはリアルタイムデータ統計だけでなく、運用やプロダクト戦略の効果を確認するために、履歴データ（昨日、さらには昨年同期間）との比較も必要としています。
ユーザーの視点から見ると、従来のストリームとバッチが独立したソリューションにはいくつかの課題があります。


人的コストが比較的高い。
ストリームとバッチは 2 つのシステムであるため、同じロジックを 2 つのチームが 2 回開発する必要があります。


データリンクの冗長性。
多くのシナリオで、ストリームとバッチのコンピューティング内容は実際には同じですが、2 つのシステムであるため、同じロジックを 2 回実行する必要があり、リソースの無駄が生じます。


データキャリバーの不整合。
これがユーザーが直面する最も重要な問題です。
2 つのシステム、2 組のオペレーター、2 組の UDF は必然的に異なる程度の誤差を生み出し、ビジネス側に大きな支障をもたらします。
これらの誤差は、人的リソースやリソースの投入だけでは簡単に解決できません。


2020 年の独身の日にリアルタイムデータのピークが 40 億件という過去最高を記録した際、Flink チームと DT チームは共同で Flink のフルリンクストリームバッチ統合に基づくデータウェアハウスアーキテクチャを構築し、Lambda アーキテクチャがもたらす一連の問題を解決しました。
ストリームバッチジョブが同一の SQL を使用することで開発効率が 3〜4 倍向上し、単一エンジンによりデータキャリバーの天然の一貫性が保証され、ストリームバッチジョブが同一クラスター上で実行されることでピークシフトとバレーフィルによりリソース効率が大幅に向上しました。


Flink のストリームバッチ統合の成功は、Flink オープンソースコミュニティの健全で活発な発展と切り離せません。
2020 年の Apache Software Foundation の年次レポートから、Flink はオープンソースコミュニティの繁栄を反映する 3 つの主要指標で優れた成績を収めていることがわかります。
ユーザーメーリングリストの活発度では Flink が 1 位、開発者のコミット数では Flink が 2 位、GitHub の訪問者数では 2 位にランクインしています。
これらのデータはビッグデータ分野だけでなく、Apache オープンソース財団の全プロジェクトを含むものです。


2020 年は Blink がコミュニティにフィードバックを始めて 2 年目にあたります。
この 2 年間で、Blink がグループ内で蓄積した経験を段階的にコミュニティに還元し、Flink を真のストリームバッチ統合プラットフォームにしてきました。
この記事を通じて、過去 2 年間に Flink が実行エンジンのストリームバッチ融合に関して行ってきたことを共有するとともに、Flink のストリームバッチ統合アーキテクチャの「前世と今生」について、Flink の旧ユーザーにも新しいユーザーにも理解を深めていただきたいと思います。


2. ストリーミングとバッチを統合する階層アーキテクチャ

一般的に、Flink のコアエンジンは主に以下の 3 つの層に分割されます。


SDK 層。
Flink の SDK には主に 2 種類あります。
1 つ目はリレーショナル SDK で、SQL/Table です。
2 つ目は物理 SDK で、DataStream です。
これら 2 種類の SDK はストリームバッチ統一されており、SQL であっても DataStream であっても、ユーザーのビジネスロジックは一度開発するだけで、ストリームとバッチの両方のシナリオで同時に使用できます。


実行エンジン層。
実行エンジンは、データ処理フローを記述する統一された DAG（Data Processing Pipeline、論理実行計画）を提供します。
ストリームタスクであってもバッチタスクであっても、ユーザーのビジネスロジックは実行前にこの DAG グラフに変換されます。
実行エンジンは、Unified DAG Scheduler を通じてこの論理 DAG を分散環境で実行されるタスクに変換します。
タスク間のデータは Shuffle を通じて転送されます。
Pluggable Unified Shuffle アーキテクチャを使用して、ストリーミングとバッチの両方の Shuffle 方式をサポートしています。


状態ストレージ。
状態ストレージ層はオペレーターの実行状態を保存する役割を担います。
ストリームジョブ向けには、オープンソースの RocksdbStatebackend、MemoryStatebackend、商用版の GemniStatebackend があり、バッチジョブ向けには、コミュニティ版で BatchStateBackend を導入しました。


この記事では主に以下の側面について共有します。


ストリームバッチ統合 DataStream：ストリームバッチ統合 DataStream を通じて Flink SDK が直面する課題をどのように解決するかを紹介します。


ストリーミングとバッチを統合する DAG Scheduler：統一された Pipeline Region メカニズムを通じてストリーミングエンジンの性能優位性を十分に引き出す方法、実行計画の動的調整によるエンジンの使いやすさの向上とシステムのリソース使用率の改善について紹介します。


ストリームバッチ統合 Shuffle アーキテクチャ：統一された Shuffle アーキテクチャを使用して、異なる Shuffle の戦略カスタマイズ要件を満たしつつ、共通要件の重複開発を回避する方法を紹介します。


ストリーミングとバッチを統合するフォールトトレランス戦略：統一されたフォールトトレランス戦略を使用して、バッチシナリオでのフォールトトレランスを満たしつつ、ストリーミングシナリオでのフォールトトレランス効果を向上させる方法を紹介します。


3. ストリームバッチ統合 DataStream

SDK の分析と課題

上記の図に示すように、現在 Flink には 3 種類の SDK が提供されています。


Table/SQL は高レベルなリレーショナル SDK で、主にデータ分析シナリオで使用され、Bounded と Unbounded の両方の入力をサポートできます。
Table/SQL は宣言型であるため、システムはユーザーに対して多くの最適化を行うことができます。
たとえば、ユーザーが指定した Schema に基づいて、Filter Push Down（述語プッシュダウン）やバイナリデータのオンデマンドデシリアライズなどの最適化を実行できます。
現在、Table/SQL は Batch と Streaming の両実行モードをサポートしています。
[1]

DataStream は物理 SDK です。
リレーショナル SDK は強力ですが、いくつかの制約もあります。
State や Timer の操作をサポートしていないこと、Optimizer のアップグレードにより、同一の SQL でも 2 つのバージョン間で物理実行計画に非互換が生じる可能性があることです。
DataStream SDK は State と Timer の次元での低レベル操作をサポートできます。
同時に、DataStream は命令型 SDK であるため、物理実行計画に対する高い「制御性」を持ち、バージョンアップグレードによる非互換性はありません。
DataStream はコミュニティで依然として大きなユーザー基盤を維持しており、たとえば、約 500 件の未クローズ DataStream issue が残っています。
DataStream は Bounded と Unbounded の両方の入力に対応したアプリケーションを記述できますが、Flink 1.12 より前では Streaming 実行モードのみをサポートしていました。


DataSet は Bounded 入力のみをサポートする物理 SDK です。
Bounded の特性に応じて一部のオペレーターを最適化しますが、EventTime や State などの操作はサポートしません。
DataSet は Flink が最初に提供した SDK ですが、リアルタイム処理とデータ分析シナリオの継続的な発展に伴い、DataStream や SQL と比較して、コミュニティでの DataSet の影響力は徐々に低下しています。


現在、Table/SQL はストリームバッチ統一シナリオに対して比較的成熟したサポートを持っていますが、Physical SDK にはまだ課題があり、主に 2 つの側面に現れています。


既存の Physical SDK では、実運用に耐えうるストリームバッチ統合アプリケーションを作成できません。
たとえば、ユーザーが Kafka のリアルタイムデータを処理するプログラムを書いた場合、同じプログラムで OSS、S3、HDFS に保存された履歴データを処理することも非常に自然です。
しかし現在、DataSet も DataStream もユーザーのこの「シンプルな」要望を満たせていません。
DataStream は Bounded Input と Unbounded Input の両方をサポートしているのに、なぜまだ問題があるのか不思議に思うかもしれません。
実際、「細部に悪魔が潜んでいる」のであり、Unified DataStream のセクションで詳しく説明します。


学習と理解のコストが比較的高い。
Flink の成長に伴い、ますます多くの新しいユーザーが Flink コミュニティに参加していますが、これらの新規ユーザーにとって、2 つの Physical SDK を学ぶ必要があります。
他のエンジンと比較して、ユーザーが入門するための学習コストが比較的高いこと、2 つの SDK のセマンティクスが異なること（たとえば、DataStream には Watermark と EventTime がありますが、DataSet にはありません）、ユーザーにとって 2 つのメカニズムを理解するハードルは小さくないこと、2 つの SDK に互換性がないため、新しいユーザーが間違った方を選ぶと大きな切り替えコストに直面することが課題です。


Unified Physical SDK

上記の Physical SDK が直面する課題を解決するため、Unified DataStream SDK を Flink の統一 Physical SDK として採用します。
この部分では主に 2 つの問題を解決します。


なぜ DataStream を Unified Physical SDK として選んだのか。


「古い」DataStream と比較して、Unified DataStream はどのような能力を提供し、ユーザーが実運用に耐えうるストリームバッチ統合アプリケーションを記述できるようにするのか。


なぜ DataSet で統一しないのか

学習と理解のコストが比較的高いという問題を解決するため、最も自然でシンプルな方法は DataStream と DataSet のいずれかを Flink 唯一の Physical SDK として選ぶことです。
では、なぜ DataSet ではなく DataStream を選んだのでしょうか。
主に 2 つの理由があります。


ユーザーメリット。
前述の分析の通り、Flink コミュニティの発展に伴い、DataSet のコミュニティでの影響力は徐々に低下しています。
DataSet を Unified Physical SDK として選ぶと、ユーザーが DataStream にこれまで行ってきた大規模な「投資」が無駄になります。
DataStream を選ぶことで、多くのユーザーが既存の DataStream への「投資」に対して追加のリターンを得られます。


開発コスト。
DataSet は古すぎて、モダンなリアルタイムコンピューティングエンジンの基本概念に対するサポートが不足しています。
EventTime、Watermark、State、Unbounded Source などがその例です。
もう 1 つのより深い理由は、既存の DataSet オペレーターの実装をストリーミングシナリオで再利用できないことです。
Join などがその例です。
DataStream の場合はそうではなく、大幅に再利用できます。
では、どのようにしてストリームとバッチの両方のシナリオで DataStream オペレーターを再利用するのでしょうか。


Unified DataStream

Flink をある程度理解しているユーザーの多くはこう尋ねるかもしれません。
DataStream は Bounded/Unbounded 入力を同時にサポートしているのに、なぜ DataStream で実運用に耐えうるストリームバッチ統合アプリケーションを記述できないと言うのか。
簡単に言うと、DataStream は元々 Unbounded シナリオ向けに設計されたため、Bounded シナリオでは効率性、ユーザビリティ、使いやすさの面で従来のバッチエンジンとまだ一定の差があるからです。
具体的には以下の 2 つの側面に現れます。


- 効率性

まず例をお見せします。
以下は、同じ規模の WordCount ジョブに対する DataStream と DataSet の性能比較グラフです。
この例からわかるように、DataSet の性能は DataStream のほぼ 5 倍です。


明らかに、DataStream でストリーミングとバッチの両方のシナリオを本番環境でサポートするには、Bounded シナリオでの DataStream の効率を大幅に向上させる必要があります。
では、なぜ DataStream は DataSet よりも効率が悪いのでしょうか。


前述の通り、DataStream の本来の主な設計目標は Unbounded シナリオで使用されることであり、Unbounded シナリオの主な特徴はデータの順序不同です。
つまり、どの DataStream オペレーターも Record が処理される順序を想定できません。
そのため、多くのオペレーターは K/V ストレージを使用してこれらの非順序データをキャッシュし、適切なタイミングで K/V ストレージからデータを取り出して処理・出力します。
一般的に、オペレーターの K/V ストレージへのアクセスには大量のシリアル化とデシリアル化が伴い、ランダムなディスク I/O も発生します。
一方、DataSet ではデータが有界であることを前提としているため、ランダムなディスク I/O アクセスを回避する最適化が可能で、シリアル化とデシリアル化の最適化も行えます。
これが DataSet で書かれた WordCount が DataStream で書かれた WordCount より 5 倍高速である主な理由です。


理由がわかったところで、DataStream のすべてのオペレーターを書き換える必要があるのでしょうか。
理論上は問題ありませんが、DataStream には書き換えが必要な大量のオペレーターがあり、Window 関連の一連のオペレーターのように比較的複雑なものもあります。
想像できる通り、すべて書き換えると作業量は膨大になります。
そこで、単一キーの BatchStateBackend を通じて、ほぼすべてのオペレーターの書き換えを回避し、同時に非常に良い結果を達成しました。


- 一貫性

Flink をある程度理解している読者であれば、DataStream で書かれた従来のアプリケーションは Streaming 実行モードを採用していることをご存知でしょう。
このモードでは、Checkpoint を通じてエンドツーエンドの Exactly Once セマンティクスを維持します。
具体的には、ジョブの Sink は、グラフ全体の（Sink 自身を含む）すべてのオペレーターがそれぞれの Snapshot を完了した後で初めて、Sink がデータを外部システムにコミットします。
これは Flink の Checkpoint メカニズムに依存した典型的な 2PC プロトコルです。


Bounded シナリオでは、Streaming を使用することも可能ですが、ユーザーにとっていくつかの問題が生じる可能性があります。


リソース消費が大きい：Streaming 方式を使用すると、すべてのリソースを同時に取得する必要があります。
場合によっては、ユーザーはそれほど多くのリソースを持っていない可能性があります。


フォールトトレランスのコストが高い：Bounded シナリオでは、一部のオペレーターは効率性の面で Snapshot 操作をサポートできない場合があり、エラーが発生するとジョブ全体を再実行する必要がある可能性があります。


そのため、Bounded シナリオでは、ユーザーはアプリケーションが Batch 実行モードを採用することを望みます。
Batch 実行モードは上記の 2 つの問題を自然に解決できるからです。
Bounded シナリオでの Batch 実行モードのサポートは比較的簡単ですが、非常に困難な問題が発生します。
既存の Sink API ではエンドツーエンドの Exactly Once セマンティクスを保証できないのです。
これは、Bounded シナリオには Checkpoint がなく、従来の Sink は Checkpoint に依存してエンドツーエンドの Exactly Once を保証していたためです。
同時に、開発者が異なるモード用の Sink に 2 つの異なる実装を開発する必要がないようにしたいと考えています。
そうしないと、Flink と他のエコシステムとの連携の利点を活かせないからです。


実際、Transactional Sink は主に以下の 4 つの問題を解決します。


何をコミットするか。

どのようにコミットするか。

どこにコミットするか。

いつコミットするか。


Flink は sink 開発者が「何をコミットするか」と「どのようにコミットするか」を提供できるようにし、システムが異なる実行モードに応じて「どこにコミットするか」と「いつコミットするか」を選択して、エンドツーエンドの Exactly Once を保証するべきです。
最終的に、新しい Unified Sink API を提案しました。
これにより、開発者は 1 つの sink を開発するだけで、Streaming と Batch の両方の実行モードで動作できます。
ここで紹介したのは主な考え方に過ぎません。
有界シナリオでのエンドツーエンドの一貫性をどのように保証するか、Hive や Iceberg などの外部エコシステムとどのように連携するかについては、まだ一定の課題が残っています。


4. ストリーミングとバッチを統合する DAG Scheduler

Unified DAG Scheduler はどのような問題を解決するのか

従来の Flink には 2 つのスケジューリングモードがありました。


1 つはストリームスケジューリングモードです。
このモードでは、Scheduler がジョブに必要なすべてのリソースを要求し、ジョブのすべてのタスクを同時にスケジュールします。
すべてのタスクはパイプライン形式で通信します。
バッチジョブもこの方式を使用でき、性能も大幅に向上します。
ただし、長時間実行されるバッチジョブの場合、このモードにはまだ一定の問題があります。
規模が比較的大きい場合に同時に多くのリソースを消費し、一部のユーザーはそれほど多くのリソースを持っていない可能性があります。
フォールトトレランスのコストが比較的高く、たとえば、エラーが発生するとジョブ全体を再実行する必要があります。


もう 1 つはバッチスケジューリングモードです。
このモードは従来のバッチエンジンに似ています。
すべてのタスクは個別にリソースを要求でき、タスク間は Batch Shuffle を通じて通信します。
この方式の利点は、フォールトトレランスのコストが比較的小さいことです。
ただし、この運用モードにもいくつかの欠点があります。
たとえば、タスク間のデータはすべてディスクを介してやり取りされるため、大量のディスク I/O が発生します。


一般的に、これら 2 つのスケジューリング方式はストリームバッチ統合シナリオの要件を基本的に満たせますが、改善の余地はまだ大きくあり、具体的には 3 つの側面に現れます。


1. 構造の不一致と高いメンテナンスコスト。
スケジューリングの本質はリソース割り当てであり、言い換えると、いつどのタスクをどこにデプロイするかを解決することです。
従来の 2 つのスケジューリングモードはリソース割り当てのタイミングと粒度に一定の差異があり、結果としてスケジューリングアーキテクチャを完全に統一できず、開発者が 2 組のロジックを保守する必要がありました。
たとえば、ストリームスケジューリングモードではリソース割り当ての粒度が物理実行計画内のすべてのタスクであるのに対し、バッチスケジューリングモードではリソース割り当ての粒度が単一のタスクです。
Scheduler がリソースを取得した際、ジョブタイプに応じて 2 通りの異なる処理ロジックに分岐する必要があります。


2. パフォーマンス。
従来のバッチスケジューリング方式ではフォールトトレランスのコストが比較的小さいものの、大量のディスク I/O が導入され、性能が最適ではないため、Flink ストリーミングエンジンの利点を活かせません。
実際、リソースが比較的に十分なシナリオでは、「ストリーム」スケジューリング方式を採用してバッチジョブを実行することで、余分なディスク I/O を回避し、ジョブの実行効率を向上できます。
特に夜間はストリーミングジョブが一定のリソースを解放するため、バッチジョブが「ストリーミング」モードで実行することが可能になります。


3. 適応性。
現在、両方のスケジューリング方式の物理実行計画は静的です。
静的に生成された物理実行計画には、チューニングの人的コストが高い、リソース使用率が低いなどの問題があります。


Pipeline Region ベースの統一スケジューリング

ストリーミングエンジンの利点を十分に活かしつつ、全グラフ同時スケジューリングの欠点を回避するため、Pipeline Region の概念を導入します。
Unified DAG Scheduler では、DAG グラフ内でタスクが Pipeline または Blocking で通信できます。
Pipeline のデータ交換方式で接続されたこれらのタスクを Pipeline Region と呼びます。
上記の概念に基づき、Flink は Pipeline Region の概念を導入しました。
ストリームジョブであってもバッチジョブであっても、Pipeline Region の粒度に従ってリソースが要求され、タスクがスケジュールされます。
注意深い読者であれば、従来の 2 つのモードは実際には Pipeline Region スケジューリングの特殊ケースであることに気づくでしょう。


たとえ「ストリーム」スケジューリングモードがリソース的に満たされたとしても、どのタスクを「ストリーム」方式でスケジュールできるのでしょうか。


一部の読者は、「ストリーム」スケジューリング方式を採用するとフォールトトレランスのコストが高くなるのではないかと心配しています。
「ストリーム」スケジューリング方式では、あるタスクでエラーが発生すると、それに接続されたすべてのタスクが失敗して再実行されるからです。


Flink では、異なるタスク間に 2 種類の接続方式があります [2]。
1 つは All-to-All 接続方式で、上流 Task がすべての下流タスクと接続されます。
もう 1 つは PointWise 接続方式で、上流 Task が一部の下流タスクのみと接続されます。


ジョブのすべてのタスクが All-to-All 方式で接続されている場合、「ストリーム」スケジューリング方式を採用すると物理トポロジー全体を同時にスケジュールする必要があるため、確かに FailOver のコストが比較的高くなる問題があります [3]。
しかし、実際のバッチジョブトポロジーでは、すべてのタスクが All-to-All エッジで接続されているわけではありません。
バッチジョブの多くの Task は PointWise エッジで接続されており、PointWise で接続されたタスクを「ストリーム」方式でスケジュールすることで、接続グラフを活用してジョブの実行効率を向上させつつ、フォールトトレランスコストを削減できます。
以下の図に示すように、フル 10T TPC-DS テストで、すべての PointWise エッジに Pipeline リンク方式を有効化することで、全体のパフォーマンスが 20% 以上向上します。


上記は Scheduler が提供する Pipeline Region を分割する 4 つの戦略のうちの 1 つに過ぎません [4]。
実際、Planner は実際の運用シナリオに応じて、どの Task が Pipeline 伝送モードを採用し、どの Task が Batch 伝送モードを採用するかをカスタマイズできます。


適応的スケジューリング

スケジューリングの本質は物理実行計画に対するリソース割り当ての意思決定プロセスです。
Pipeline Region が物理実行計画の確定問題を解決した後、ストリームジョブとバッチジョブは Pipeline Region の粒度で統一的にスケジュールできます。
バッチジョブの物理実行計画を静的に生成することにはいくつかの問題があります [5]。


人的コストが高い。
バッチジョブの場合、物理実行計画の各ステージの同時実行理論的には統計情報に基づいて推論できますが、大量の UDF や統計情報の不足により、静的な決定結果が大幅に不正確になる可能性があります。
ビジネス運用の SLA を確保するため、大規模プロモーション期間中にはビジネス担当者がプロモーションのトラフィック見積もりに基づいて高品質なバッチジョブの同時実行数を手動で調整する必要があり、この調整プロセス全体がビジネス担当者の手動作業を必要とし、人的コストが比較的高い上、それでも誤判断が発生し、ユーザー SLA を満たせない可能性があります。


リソース使用率が低い。
同時実行の手動設定コストが高いため、すべてのジョブに対して手動で同時実行数を設定することは不可能です。
中低優先度のジョブについては、ビジネス担当者はいくつかのデフォルト値を同時実行数として選択しますが、ほとんどの場合これらのデフォルト値は大きすぎてリソースの無駄になります。
また、高優先度ジョブは手動で同時実行数を設定できますが、設定方法が煩雑なため、大規模プロモーション終了後にトラフィックが低下しても、ビジネス側はプロモーション期間中の設定をそのまま使用し続け、これも大量のリソース浪費を引き起こします。


安定性が低い。
リソースの無駄は最終的にリソースの過剰予約につながります。
現在、ほとんどのバッチジョブはストリーミングジョブクラスターと混在しており、具体的には、リクエストされたリソースはすべて非保証リソースです。
リソースが逼迫したりマシンにホットスポットが発生したりすると、これらの非保証リソースが最初に調整されます。


静的に生成された物理実行計画のこれらの問題を解決するため、バッチジョブに対して適応的スケジューリング機能を導入しました [6]。
従来の静的な物理実行計画と比較して、この機能を使用することでユーザーのリソース使用率を大幅に向上できます。
Adaptive Scheduler はある JobVertex の上流 JobVertex の実行状態に基づいて、現在の JobVertex の同時実行数を動的に決定できます。
将来的には、上流 JobVertex が生成したデータに基づいて、下流でどのオペレーターを使用するかを動的に決定することもできます。


5. ストリームとバッチを統合する Shuffle アーキテクチャ

Flink はストリームバッチ統合プラットフォームであり、異なる実行モードのためにストリーミングとバッチの 2 種類の Shuffle を提供しています。
Streaming Shuffle と Batch Shuffle は具体的な戦略に一定の差異がありますが、本質的にはデータの再パーティショニングであるため、異なる Shuffle 間には一定の共通性があります。
そのため、統一された Shuffle アーキテクチャを提供することで、異なる Shuffle 戦略のカスタマイズ要件を満たしつつ、共通要件の重複開発を回避することが目標です。


一般的に、Shuffle アーキテクチャは以下の図に示す 4 つのレベルに分割できます。
各レベルでストリームとバッチの Shuffle 要件にはいくらかの違いがあり、同時に多くの共通性もあります。
以下に簡単な分析を行います。


Streaming Batch Shuffle の違い

バッチジョブとストリーミングジョブでは Shuffle に対する要件が異なることは周知の通りで、以下の 3 つの側面に現れます。


1. Shuffle データのライフサイクル。
ストリームジョブの Shuffle データのライフサイクルは Task のライフサイクルとほぼ一致しますが、バッチジョブの Shuffle データは Task のライフサイクルから切り離されています。


2. Shuffle データの保存媒体。
ストリームジョブの Shuffle データはライフサイクルが比較的短いため、メモリに保存できます。
一方、バッチジョブの Shuffle データのライフサイクルには一定の不確実性があるため、ディスクに保存する必要があります。


3. Shuffle デプロイ方式 [7]。
Shuffle サービスとコンピューティングノードを一緒にデプロイすることはストリームジョブに有利です。
不要なネットワークオーバーヘッドを削減し、レイテンシを低減できるからです。
しかし、バッチジョブにとっては、このデプロイ方式はリソース使用率、性能、安定性の面で一定の問題があります [8]。

Streaming Batch Shuffle の共通性

バッチジョブとストリームジョブの Shuffle には違いと共通性があります。共通性は主に以下に現れます。

1. データのメタ管理。いわゆる Shuffle Meta とは、論理データ分割からデータ物理位置へのマッピングを指します。ストリームであれバッチであれ、正常時には Meta からデータの読み書きの物理位置を特定する必要があり、異常時にはフォールトトレランスのコストを削減するため、通常 Shuffle Meta データは永続化されます。
2. データ転送。論理的には、ストリームジョブとバッチジョブの Shuffle はデータの再パーティショニング/再分配です。分散システムでは、データの再パーティショニングにはスレッド、プロセス、マシン間のデータ転送が伴います。

ストリームバッチ統合 Shuffle アーキテクチャ

Unified Shuffle アーキテクチャは 3 つのコンポーネントに抽象化されています [9]：Shuffle Master、Shuffle Reader、Shuffle Writer です。Flink はこれら 3 つのコンポーネントとの相互作用を通じて、オペレーター間のデータ再パーティショニングを完了します。異なる Shuffle プラグインの具体的な戦略の違いも、これら 3 つのコンポーネントを通じて満たせます。

Shuffle Master：リソースの申請と解放。つまり、プラグインはフレームワークに対してリソースの要求/解放方法を通知する必要があり、Flink がいつ呼び出すかを決定します。
上流オペレーターは Shuffle Writer を使用して Shuffle Service にデータを書き込みます。ストリーミング Shuffle はメモリにデータを書き込み、外部/リモート Batch Shuffle は外部ストレージにデータを書き込みます。

Shuffle Reader の下流オペレーターは Reader を通じて Shuffle データを読み取れます。
同時に、ストリームバッチ Shuffle の共通性（メタ管理、データ転送、サービスデプロイ）[10] に対してもアーキテクチャレベルのサポートを提供し、複雑なコンポーネントの重複開発を回避しています。効率的で安定したデータ転送は分散システムの最も複雑なサブシステムの 1 つです。たとえば、バックプレッシャー、データ圧縮、メモリゼロコピーなどの問題が転送時に解決される必要があります。新しいアーキテクチャでは、これらを一度だけ開発すればよく、同時にストリームとバッチの両方のシナリオで使用されるため、開発とメンテナンスのコストが大幅に削減されます。

6. ストリーミングとバッチを統合するフォールトトレランス戦略

Flink の従来のフォールトトレランス戦略はチェックポイントベースです。具体的には、単一 Task の失敗であれ JobMaster の失敗であれ、Flink は最新のチェックポイントに基づいてジョブ全体を再起動します。この戦略には最適化の余地があるものの、ストリーミングシナリオでは一般的に受け入れられています。現在、Flink の Batch 実行モードではチェックポイント [11] が有効化されておらず、エラーが発生するとジョブ全体を最初から実行する必要があることを意味します。

従来の戦略は理論上最終的に正しい結果を生成できるものの、このフォールトトレランス戦略のコストはほとんどのユーザーにとって受け入れがたいものです。これらの問題を解決するため、Task と JM のフォールトトレランスに対してそれぞれ対応する改善を行いました。

Pipeline Region Failover

Batch 実行モードには定期的な Checkpoint がありませんが、Batch 実行モードでは Flink は Blocking Shuffle を通じた Task 間通信を許可しています。Blocking Shuffle を読み取る Task が失敗した場合、Blocking Shuffle にはその Task に必要なすべてのデータが保存されているため、その Task とそれに Pipeline Shuffle で接続されたすべて下流タスクのみを再起動すればよく、ジョブ全体を再起動する必要はありません。

一般的に、Pipeline Region Failover 戦略は Scheduler が通常のスケジューリングを行う際のロジックと同じです。DAG をいくつかの Pipeline Shuffle で接続された Pipeline Region に分割し、ある Task で FailOver が発生するたびに、その Task が属する Region のみを再起動すれば十分です。

JM Failover

JM はジョブのコントロールセンターであり、ジョブのさまざまな実行状態を含んでいます。Flink はこれらの状態を使用してタスクのスケジューリングとデプロイを行います。JM にエラーが発生すると、これらすべての状態が失われます。この情報なしでは、すべてのワーカーノードが正常であっても、新しい JM は元のジョブのスケジューリングを継続できません。たとえば、タスクの終了情報が既に失われているため、あるタスクが終了した後、新しい JM は既存の状態が下流タスクのスケジューリング条件（すべての入力データが生成されていること）を満たしているかを判断できません。

上記の分析から、JM Failover の鍵は JM の「記憶を復元」する方法にあることがわかります。VVR [12] では、Operation Log ベースのメカニズムを通じて JM の重要な状態を復元します。

注意深い読者は気づいたかもしれませんが、これら 2 つの改善策の出発点はバッチシナリオ向けですが、実際にはストリーミングジョブにも有効です。上記は 2 つのフォールトトレランス戦略の考え方の簡単な紹介に過ぎません。実際にはまだ検討すべきことが多くあります。たとえば、Blocking 上流データの損失にどのように対応すべきか、JM のどのような重要な状態を復元する必要があるのかといった点です。

7. 将来の展望

現在よりもさらに高速で安定したユーザー体験を提供するため、次世代ストリーミングアーキテクチャの研究開発に着手しました。Flink はストリームバッチ統合シナリオでますます多くのユーザーから認められていますが、業界にはまだ学ぶべき多くの優れた従来のビッグデータシステムが存在することも認識しています。最後に、完璧なユーザー体験を持つストリームバッチ統合ビッグデータコンピューティングエンジンの構築に参加してくださる興味のある仲間をお待ちしています。