Flink Remote Shuffle open source

1. Flink Remote Shuffle が必要な理由

1.1 背景

Flink Remote Shuffle の提案と実装は、ストリームバッチ統合とクラウドネイティブへの高まりに対応するために行われました。

リアルタイム処理はユーザー体験を大幅に向上させ、市場でのプロダクト競争力を高めることができるため、ますます多くのユーザーのビジネスシナリオにリアルタイム処理とオフライン処理の両方の要件が含まれるようになっています。ストリーム処理とバッチ処理に異なるフレームワークを使用すると、フレームワークの学習、コード開発、オンライン運用保守において多くの不便が生じます。同時に、多くのアプリケーションシナリオでは、リアルタイム処理が遅延データ（たとえば、ユーザーがコメントを入力するまで長時間待つ場合があります）やビジネスロジックのアップグレードによる制約を受けるため、データ修正にはオフラインタスクを使用する必要があり、2 つの異なるフレームワークで 2 つのコードロジックを書くと計算結果の整合性に問題が生じやすくなります。

これらの問題に対応するため、Flink はストリームバッチ統合データモデルを提案しました。このモデルは 1 セットの API でリアルタイムデータとオフラインデータの両方の処理を完遂します。この目標をサポートするため、Flink はストリームバッチ統一の DataStream API [1] + Table/SQL API [2] + Connector [3][4] を設計・実装し、実行層ではストリームバッチ統合スケジューリング [5] とバッチ処理に最適化された Batch 実行モード [6] をサポートしています。Batch モードをサポートするためには、Flink は効率的かつ安定したブロッキングシャッフルも実装できる必要があります。Flink の組み込みブロッキングシャッフルは、上流の TaskManager が上流終了後も下流のためにデータ読み取りサービスを提供し続ける方式です。これにより TaskManager が直ちに解放されず、リソース使用率が低下し、シャッフルサービスの安定性がタスク実行の安定性に影響を受けるという問題があります。

一方、クラウドネイティブはオフラインとオンラインの混合処理によるクラスターリソース使用率の向上、統合運用管理インターフェイスによる運用保守コストの削減、リソースの動的オーケストレーションによるジョブの自動スケーリングをより良くサポートできるため、ますます多くのユーザーが Kubernetes を使ってクラスターリソースを管理し始めています。Flink はクラウドネイティブに積極的に対応しており、Kubernetes のネイティブサポート [7][8] の提供に加え、リソース量に応じて動的にスケーリングする Adaptive Scheduler [9] を提供し、State ストレージとコンピューティングの分離 [10] を段階的に推進しています。Batch モードをクラウドネイティブ環境でより良くサポートするためには、ローカルディスクの最大の利用者であるシャッフル処理において、ブロッキングシャッフルのストレージとコンピューティングの分離を実現し、ローカルディスクの占有を減らし、コンピューティングリソースとストレージリソースが互いに結合されないようにすることも、解決すべき重要な課題です。

したがって、ストリームバッチ統合とクラウドネイティブをより良くサポートするためには、独立したシャッフルサービスを使ってタスク間のデータ転送を実現することが不可欠です。

1.2 Flink Remote Shuffle の利点

Flink Remote Shuffle は上記の考え方に基づいて設計・実装されており、以下の重要な機能を備えています。

ストレージとコンピューティングの分離：ストレージとコンピューティングの分離により、コンピューティングリソースとストレージリソースの独立したスケーリングが可能になります。コンピューティング完了後、コンピューティングリソースは直ちに解放できます。シャッフルの安定性はコンピューティングの安定性に影響されなくなります。
複数のデプロイモードのサポート：Kubernetes、YARN、および Standalone 環境でのデプロイをサポートします。
Flink の Credit-based フロー制御と同様のメカニズムを採用し、ゼロコピーデータ転送を実現し、マネージドメモリを最大限に活用して OOM を回避し、システムの安定性とパフォーマンスを向上させています。

負荷分散、ディスク IO 最適化、データ圧縮、接続多重化、小パケットマージなど、多くの最適化を実現し、優れたパフォーマンスと安定性を達成しています。

シャッフルデータの整合性検証をサポートし、シャッフルプロセスや物理ノードの再起動にも耐えられます。
FLIP-187: Flink Adaptive Batch Job Scheduler [11] と組み合わせることで、演算子の同時実行レベルを動的に決定するなどの動的実行最適化をサポートできます。

1.3 本番運用の実践
2020 年の独身の日から、阿里内の多くのコアタスクが Flink ベースのストリームバッチ統合処理リンクを採用し始めました。これは業界初の大規模なストリームバッチ統合の本番運用です。ストリームバッチ統合処理技術により、Tmall Marketing Engine などのシナリオにおけるストリームバッチ処理の整合性問題を解決し、データレポート開発の効率を 4 倍から 10 倍に向上させました。夜間のピークシフトとバレーフィルにより、リソースコストをさらに削減しています。

ストリームバッチ統合技術の重要な構成要素として、Flink Remote Shuffle はサービス開始以来、最大クラスター規模が 1,000 台以上に達しています。Tmall Marketing Engine、Tmall International などの複数のビジネス側を安定的にサポートし、運用規模は PB レベルを超えており、システムの安定性とパフォーマンスを十分に実証しています。

2. Flink Remote Shuffle の設計と実装

2.1 Flink Remote Shuffle の全体アーキテクチャ

Flink Remote Shuffle は Flink の統一プラグインシャッフルインターフェイスに基づいて実装されています。ストリームバッチ統合データ処理プラットフォームとして、Flink はネットワークベースのオンラインパイプラインシャッフル、TaskManager ベースのブロッキングシャッフル、リモートサービスベースのリモートシャッフルなど、さまざまなシナリオに適した多様なシャッフル戦略に対応できます。

これらのシャッフル戦略は転送方法とストレージメディアの点で大きく異なりますが、データセットのライフサイクル、メタデータ管理、下流タスクへの通知、データ分散戦略に関しては多くの共通要件を持っています。さまざまなタイプのシャッフルに統一的なサポートを提供し、Flink Remote Shuffle を含む新しいシャッフル戦略の実装を簡素化するため、Flink はプラグインシャッフルアーキテクチャ [12] を導入しました。

下図に示すように、シャッフルプラグインは主に 2 つの部分で構成されています。JobMaster 側でリソースの申請と解放を担当する ShuffleMaster、および TaskManager 側で実際のデータ読み書きを担当する InputGate と ResultPartition です。スケジューラが ShuffleMaster を通じてリソースを申請した後、PartitionTracker によって管理され、上流タスクと下流タスクが起動する際にシャッフルリソースの記述子を携行して、データ出力と読み取りの位置を記述します。

Flink の統一プラグインシャッフルインターフェイスに基づいて、Flink Remote Shuffle は独立したクラスターを通じてデータシャッフルサービスを提供します。クラスターは標準的なマスター/スレーブ構造を採用しており、ShuffleManager がクラスターのマスターノードとして、Worker ノードの管理、シャッフルデータセットの割り当てと管理を担当します。ShuffleWorker はクラスターのスレーブノードとして、データセットの実際の読み取り、書き込み、クリーニングを担当します。

上流タスクが起動すると、Flink スケジューラは RemoteShuffleMaster プラグインを通じて ShuffleManager にリソースを申請し、ShuffleManager はデータセットのタイプと各 Worker の負荷に基づいて適切な Worker を選択してサービスを提供します。スケジューラが対応するシャッフルリソース記述子を取得すると、上流タスクを起動する際にその記述子を携行します。上流タスクは記述子に記録された ShuffleWorker アドレスに基づいて、対応する ShuffleWorker にデータを送信し永続ストレージに保存します。一方、下流タスクが起動すると、記述子に記録されたアドレスに基づいて対応する ShuffleWorker からデータを読み取り、データ転送プロセス全体を完了します。

長期稼働サービスとして、システムのフォールトトレランスと自己修復機能は極めて重要です。Flink Remote Shuffle はハートビートなどのメカニズムを通じて ShuffleWorker と ShuffleManager を監視し、ハートビートタイムアウトや IO 障害などの異常が発生した際にデータセットの状態を削除・同期することで、クラスター全体の状態の最終的な一貫性を維持します。例外処理の詳細については、Flink Remote Shuffle の関連ドキュメント [13] を参照してください。

2.2 データシャッフルプロトコルと最適化
データリモートシャッフルは、読み取りと書き込みの 2 つのフェーズに分けられます。データ書き込みフェーズでは、上流のコンピューティングタスクの出力データがリモートの ShuffleWorker に書き込まれます。データ読み取りフェーズでは、下流のコンピューティングタスクがリモートの ShuffleWorker から上流のコンピューティングタスクの出力を読み取って処理します。データシャッフルプロトコルは、このプロセスにおけるデータ型、粒度、制約、および処理フローを定義します。全体として、データの読み書きプロセスは以下のようになります。

データ読み書きの全プロセスで、データ圧縮、フロー制御、データコピーの削減、マネージドメモリの活用など、さまざまな最適化が実装されています。

Credit-based フロー制御：フロー制御はプロデューサー・コンシューマーモデルで考慮すべき重要な課題であり、コンシューマーの処理が遅くてもデータが無限に蓄積されないようにすることを目的としています。Flink Remote Shuffle は Flink と同様の Credit-based フロー制御メカニズムを採用しています。つまり、データ送信側はデータ受信側がデータを受け取るのに十分なバッファーを持っている場合のみデータを送信します。データ受信側はデータを継続的に処理する過程で、解放されたバッファーを送信側にフィードバックして新しいデータを送信し続け、この往復により TCP のスライドウィンドウメカニズムに似たストリーミングデータ転送を実現します。Credit-based フロー制御メカニズムは、下流の受信バッファーが不足している際の無効なディスク書き込みを回避でき、TCP 接続を多重化しているシナリオでも 1 つの論理リンクの輻輳が他の論理リンクに影響を与えないようにすることができます。このメカニズムに興味がある方は、Flink のブログ [14] を参照してください。

データ圧縮：データ圧縮はシンプルで効果的な最適化手法であり、その効果は広く実証されているため必須の施策です。Flink Remote Shuffle でもデータ圧縮を実装しています。具体的には、プロデューサーがリモート ShuffleWorker にデータを書き出す前に圧縮を行い、コンシューマーがリモート ShuffleWorker からデータを読み取った後に解凍します。これにより、ネットワーク IO とファイル IO の両方の削減を同時に実現し、ネットワーク帯域幅とディスクストレージの占有を減らしながら、IO 効率を向上させます。
データコピーの削減：ネットワーク IO とファイル IO を実行する際、Flink はダイレクトメモリを最大限に活用することで、Java ヒープメモリのコピーを減らし、効率を向上させています。また、ダイレクトメモリの動的申請を削減することで、安定性の向上にもつながります。

マネージドメモリの使用：シャッフルデータ転送とファイル IO で使用される大きなメモリブロックに対して、Flink Remote Shuffle は事前に申請されたマネージドメモリを使用します。つまり、メモリを事前に申請してメモリプールを確立し、その後のメモリの申請と解放はすべてそのメモリプール内で行われます。これにより、動的なメモリ申請と解放のオーバーヘッド（システムコールと GC）を削減し、さらに重要なことに、OOM 問題の回避に役立ち、システムの安定性を大幅に向上させます。

TCP 接続多重化：同じ Flink コンピューティングノードから同じリモート ShuffleWorker へのデータ読み取りまたは書き込み接続に対して、同じ物理 TCP 接続を再利用することで、ネットワーク接続数を削減し、データ読み書きの安定性を向上させています。

2.3 ストレージとファイル IO の最適化

ディスクベースのシャッフル、特にメカニカルハードディスクでは、ファイル IO が重要なボトルネックとなり、ファイル IO の最適化は大きな効果を発揮します。

前述のデータ圧縮に加えて、広く使われている技術的手法として、小ファイルや小データブロックをマージしてファイルのシーケンシャル読み書きを増やし、過度なランダム読み書きを回避することでファイル IO パフォーマンスを最適化する手法があります。リモートではないコンピューティングノード間の直接シャッフルでは、Spark を含むシステムが小データブロックを大きなブロックにマージする最適化を既に実現しています。

リモートシャッフルシステムのデータマージ方式については、調査によると、Microsoft、LinkedIn、Quantcast による論文 Sailfish [15] で最初に提案され、その後 Princeton と Facebook の Riffle [16]、Facebook の Cosco [17]、LinkedIn の Magnet [18]、Alibaba EMR の Spark Remote Shuffle [19] いずれも同様の最適化アイデアを実装しています。すなわち、異なる上流コンピューティングタスクから同じ下流コンピューティングタスクに送信されるシャッフルデータを同じリモートシャッフルサービスノードでマージし、下流のコンピューティングタスクはこれらのリモートシャッフルサービスノードからマージ済みデータを直接プルできます。

この最適化アイデアに加えて、Flink のコンピューティングノード間の直接シャッフル実装において、Sort-Spill に IO スケジューリングを加えた別の最適化アプローチを提案しました。簡単に説明すると、コンピューティングタスクの出力データがメモリバッファーを満たした後、データをソート（Sort）し、ソート済みデータをファイルに書き出し（Spill）、書き込み中に複数のファイルを作成するのではなく常に同じファイルにデータを追加し、データ読み取り時にはファイルのオフセット順にデータを読み取るようスケジューリングして読み取りリクエストを満たします。最適な状況では、完全なシーケンシャル読み取りを実現できます。下図は基本的なストレージ構造と IO スケジューリングプロセスを示しています。詳細については、Flink ブログ [20] またはその中国語版 [21] を参照してください。


どちらの方式にもそれぞれのメリットとデメリットがあります。

フォールトトレランスの観点では、データマージ方式はデータ損失への耐性が低くなります。同じファイルにすべての並列コンピューティングタスクからマージされたデータが含まれているため、ファイルが 1 つ失われるとすべての並列プロデューサーを再実行する必要があり、そのオーバーヘッドは膨大です。そのため、このオーバーヘッドを回避するにはバックアップを使用する必要があるかもしれませんが、バックアップは追加のファイル IO（パフォーマンス低下）とストレージ容量を意味します。IO スケジューリング方式では、データの破損や損失に対して失われたデータのみを再生成すれば済みます。さらに、プロデューサータスクの障害処理に関しても、データマージ方式はより複雑です。失敗したデータセグメントのクリーニングやマーク付けを行い、読み取り時にこれらのデータをスキップするか、読み取り時に重複排除してスキップする必要があります。IO スケジューリング方式では、失敗したプロデューサーが生成したデータファイルを破棄するだけで済みます。

パフォーマンスの観点では、一般的に両方の方式で良好なファイル IO スループットを達成できます。ただし、特殊な状況では IO スケジューリング方式にもいくつかの欠点があります。たとえば、IO スケジューリングはコンシューマーのコンピューティングタスクのデータリクエストに依存するため、データプルのタイミングがずれるとデータのシーケンシャル読み取りに影響し、ファイル IO パフォーマンスが低下します。また、データ自体をソートする必要がある場合、ソート対象のデータが同じファイル内にあるため、データマージ方式の方が有利です。同様に、分散ファイルシステムなどの外部システムにデータを書き込む必要がある場合も、これらの外部システムでは IO スケジューリングの最適化が容易ではないため、データマージ方式の方が有利です。

ファイル数の観点では、データマージ方式のファイル数はコンシューマータスク数に等しく、IO スケジューリング方式のファイル数はプロデューサータスク数に等しくなります。

Flink Remote Shuffle の抽象化は、いかなる最適化戦略も排除しません。実際、Flink Remote Shuffle は Map-Reduce セマンティクスを理解する中間データストレージサービスと見なすことができます。基本的なデータストレージユニットはデータパーティションです。データパーティションには MapPartition と ReducePartition の 2 つの基本タイプがあります。MapPartition に含まれるデータは単一の上流コンピューティングタスクによって生成され、複数の下流コンピューティングタスクによって消費される可能性があります。以下の図は MapPartition の生成と消費を示しています。

ReducePartition は複数の上流コンピューティングタスクの出力をマージして生成され、単一の下流コンピューティングタスクによって消費されます。以下の図は ReducePartition の生成と消費を示しています。

3. デプロイ、利用、評価

3.1 マルチ環境デプロイと運用保守

複数の環境でのデプロイをサポートし、個別のデプロイ要件を満たすことは重要な機能です。具体的には、Flink Remote Shuffle は Kubernetes、YARN、および Standalone の 3 つのデプロイモードをサポートしており、ほとんどのユーザーのデプロイ環境要件を満たすことができます。各デプロイモードでは、利便性のあるスクリプトとテンプレートが用意されています。詳細な情報については、ドキュメントを参照してください。Kubernetes モードデプロイ [22]、YARN モードデプロイ [23]、および Standalone モードデプロイ [24]。なお、Kubernetes モードと YARN モードのデプロイではマスターノード（ShuffleManager）の高可用性を実現しており、Standalone モードデプロイでのマスターノード高可用性は今後のバージョンでサポートされる予定です。

さらに、Flink Remote Shuffle のメトリックシステムは、システム全体の実行状況を監視するための重要なモニタリング指標をいくつか提供しています。アクティブノード数、ジョブ総数、各ノードの利用可能バッファー数、データパーティション数、ネットワーク接続数、ネットワークスループット、JVM 指標などの情報が含まれ、今後もモニタリング指標を追加して運用保守作業を支援していく予定です。各プロセス（ShuffleManager と ShuffleWorker）のメトリックサービスに直接アクセスして、対応する指標データをクエリできます。詳細については、ユーザードキュメント [25] を参照してください。今後は Prometheus などの外部システムへのメトリック主動報告機能も提供される予定です。

基本的に、Flink Remote Shuffle のデプロイと運用保守は比較的シンプルです。今後もデプロイと運用保守の体験を継続的に改善し、情報収集と問題特定の簡素化、自動化の向上、運用保守コストの削減を進めていきます。

3.2 マルチバージョン互換性

リモートシャッフルシステムはクライアントとサーバーの 2 つの部分に分かれているため、サーバーは独立クラスターとして稼働し、クライアントは Flink クラスター上で Flink ジョブがリモートシャッフルサービスにアクセスするためのエージェントとして稼働します。デプロイモードでは、複数のユーザーが異なる Flink クラスターから同じシャッフルサービスセットにアクセスする可能性があるため、マルチバージョン互換性はユーザーがより関心を寄せる課題です。シャッフルサービス自体のバージョンは新機能や最適化により継続的にアップグレードされます。クライアントとサーバー間に互換性がない場合、最も簡単な方法は異なるユーザーのクライアントも一緒にアップグレードすることですが、これにはユーザーの協力が必要であり、常に実現可能とは限りません。

バージョン間の互換性を最大限に保証することが理想的です。これを実現するため、Flink Remote Shuffle も多くの取り組みを行っています。

バージョン情報と予約フィールド：すべてのプロトコルメッセージにバージョン情報と予約フィールドを追加し、後続のプロトコルフィールド変更時に互換性を維持しやすくしています。
ストレージフォーマットバージョンの追加：保存データにストレージフォーマットバージョンを保持し、新バージョンのシャッフルストレージノードが古いデータを直接引き継げるようにすることで、データ再生成のオーバーヘッドを回避しています。
バージョン別の処理：異なるバージョンに対して異なる処理を行うことで、新バージョンが旧バージョンのロジックと互換性を保ち、サーバー側で旧バージョンクライアントの利用状況を監視することもできます。
互換バージョンのサービス検出：クライアントのサービス検出により、複数バージョンのシャッフルサービスを同時実行でき、常に自身のバージョンと互換性のあるサービスを探します。
これらの取り組みにより、異なるバージョン間の完全な互換性を実現し、不要な問題を回避できることを目指しています。ただし、新バージョンの新機能や最適化をより多く活用するには、クライアントバージョンのアップグレードが必要です。

3.3 安定性とパフォーマンスの評価

本番運用では、Flink Remote Shuffle は優れた安定性とパフォーマンスを発揮しています。これは多くのパフォーマンスと安定性の最適化に支えられています。

安定性を向上させる設計と最適化には以下が含まれます。ストレージとコンピューティングの分離により、シャッフルの安定性はコンピューティングの安定性に影響されなくなります。Credit-based フロー制御は、コンシューマーの処理能力に応じてデータを送信し、コンシューマーが圧迫されないようにします。接続多重化、小パケットマージ、主動的なネットワーク接続ヘルスチェックなどの最適化はネットワークの安定性を向上させます。マネージドメモリの最大限活用により OOM の可能性を大幅に回避します。データ検証により、システムはプロセスや物理ノードの再起動に耐えられます。

パフォーマンスの観点では、データ圧縮、負荷分散、ファイル IO 最適化がいずれもデータシャッフルのパフォーマンスを大幅に向上させています。データ量が少ない場合、ほとんどのシャッフルデータはオペレーティングシステムのキャッシュに格納されるため、Flink Remote Shuffle のパフォーマンスはコンピューティングノード間の直接シャッフルとほぼ同等で、大きな差はありません。データ量が多い場合、ShuffleManager ノードの負荷分散や個々の ShuffleWorker ノードが物理マシン全体の IO を一元管理する集中型意思決定のおかげで、Flink Remote Shuffle のパフォーマンスはさらに優れています。以下のスクリーンショットは、TPC-DS q78 ジョブ実行中の Flink Remote Shuffle のディスク IO 情報を示しています。

図から、sdd、sde、sdf、sdg、sdi、sdk の各ディスクを使用しており、ディスクスループットは比較的高い水準を維持していることが分かります。今後も継続的に最適化を進めていきます。