Alibaba Cloud RemoteShuffleService new features: AQE and flow control

RSS の AQE サポート

AQE の概要

Adaptive Query Execution (AQE) は Spark3 の重要な機能です [2]。ランタイム統計を収集することで後続の実行計画を動的に調整し、オプティマイザーが統計を正確に予測できないために最適な実行計画が生成されない問題を解決します。AQE には主にパーティション統合、結合ストラテジー切替、スキュー結合最適化の 3 つの最適化シナリオがあります。これら 3 つのシナリオはいずれもシャッフルフレームワークの能力に対する新たな要件を提示しています。

パーティションマージ

パーティションマージの目的は、リデューサーが処理するデータ量を適切かつ可能な限り均一にすることです。まず、マッパーがパーティション数に応じてシャッフルライトを実行します。AQE フレームワークは各パーティションのサイズを集計し、連続する複数のパーティションのデータ量が比較的少ない場合、それらのパーティションを 1 つに統合して 1 つのリデューサーに処理させます。そのプロセスを以下に示します。

上の図から、最適化された Reducer2 は、元々 Reducer2〜4 に属していたデータを読み取る必要があることがわかります。シャッフルフレームワークに対する要件は、ShuffleReader が範囲パーティションをサポートする必要があるということです。

結合ストラテジー切替

結合ストラテジー切替の目的は、不正確な統計推定によりオプティマイザーがソートマージ結合またはシャッフルハッシュ結合を誤って選択したことを修正することです。具体的には、結合対象の 2 つのテーブルのシャッフルライトが完了した後、AQE フレームワークが実際のサイズを算出し、小テーブルがブロードキャスト結合の条件を満たしていることが判明した場合、小テーブルをブロードキャストして大テーブルのローカルシャッフルデータと結合します。そのプロセスは以下のとおりです。

結合ストラテジー切替には 2 つの最適化があります。1. ブロードキャスト結合への書き換え、2. 大テーブルのデータを LocalShuffleReader を介してローカルで直接読み取ることです。シャッフルフレームワークに対する第 2 の新たな要件は、ローカルリードのサポートです。

スキュー結合最適化

スキュー結合最適化の目的は、より多くのリデューサーにスキューパーティションを処理させることで、ロングテールを回避することです。具体的には、シャッフルライト完了後、AQE フレームワークが各パーティションのサイズを集計し、特定のルールに基づいてスキューの有無を判定します。スキューが存在する場合、パーティションを複数の分割に分割し、各分割をもう一方のテーブルの対応するパーティションと結合します。以下に示します。

パーティション分割は、MapId の順序でシャッフル出力サイズを累積し、累積値がしきい値を超えたときに分割をトリガーします。シャッフルフレームワークに対する新たな要件は、ShuffleReader が範囲 MapId をサポートする必要があることです。パーティション統合と範囲パーティションの要件を総合的に最適化します。

RSS アーキテクチャの概要

RSS のコア設計は Push シャッフル + パーティションデータ集約です。つまり、異なるマッパーが同じパーティションに属するデータを同じワーカーにプッシュして集約し、リデューサーは集約されたファイルを直接読み取ります。下図に示します。

コア設計に加え、RSS はマルチレプリカ、フルリンクフォールトトレランス、マスター HA、ディスクフォールトトレランス、アダプティブプッシャー、ローリングアップグレードなどの機能も実装しています。詳細は [1] を参照してください。

RSS のパーティションマージサポート

パーティションマージのシャッフルフレームワークに対する要件は、範囲パーティションのサポートです。RSS では各パーティションが 1 つのファイルに対応するため、自然にサポートされます。下図に示します。

RSS の結合ストラテジー切替サポート

結合ストラテジー切替のシャッフルフレームワークに対する要件は、LocalShuffleReader のサポートです。RSS はリモート属性を持つため、データは RSS クラスターに保存され、RSS とコンピューティングクラスターが混在する構成の場合にのみローカルに存在します。そのため、ローカルリードは現時点ではサポートされていません (混在シナリオは今後最適化およびサポートの予定です)。注意点として、ローカルリードはサポートされていませんが、結合の書き換えには影響しません。RSS は結合書き換えの最適化をサポートしています。下図に示します。

RSS のスキュー結合最適化サポート

AQE の 3 つのシナリオの中で、RSS のスキュー結合最適化サポートが最も難しいポイントです。RSS のコア設計はパーティションデータ集約であり、シャッフルリードのランダム読み取りをシーケンシャル読み取りに変換することで、パフォーマンスと安定性の向上を目指しています。複数のマッパーが同時に RSS ワーカーにプッシュし、RSS はメモリ内集約後にフラッシュされるため、パーティションファイル内の異なるマッパーからのデータは順序不同になります。下図に示します。

スキュー結合最適化では、範囲マップの読み取り (例：Map1〜2 のデータ) が必要です。一般的な方法は 2 つあります。

1. ファイル全体を読み取り、範囲外のデータを破棄する。

2. インデックスファイルをインポートし、各ブロックの位置と MapId を記録し、範囲内のデータのみを読み取る。

これら 2 つの方法の問題点は明白です。方法 1 は大量の冗長なディスク読み取りを引き起こします。方法 2 は本質的にランダム読み取りに後退し、RSS のコアアドバンテージを失い、インデックスファイルの作成が一般的なオーバーヘッドとなります。スキューのないデータに対してもです (シャッフルライト中にスキューの有無を正確に予測することは困難です)。

上記 2 つの問題を解決するため、新しい設計を提案します。アクティブスプリット + Sort On Read です。

アクティブスプリット

スキューパーティションのおおよそのサイズは非常に大きく、極端な場合はディスクが直接破損する可能性があります。スキューのないシナリオでも、大きなパーティションが発生する確率は依然として低くありません。したがって、ディスク負荷分散の観点から、パーティションファイルのサイズを監視し、アクティブな分割を実行することは非常に重要です (デフォルトのしきい値は 256 MB)。

分割が発生すると、RSS は現在のパーティションに対して新しいワーカーペア (プライマリレプリカ) を再割り当てし、後続のデータは新しいワーカーにプッシュされます。Split が実行中のマッパーに影響を与えないようにするため、ソフトスプリットの方法を提案します。つまり、Split がトリガーされたとき、RSS は非同期で新しいワーカーを準備し、準備完了後にマッパーの PartitionLocation 情報をホットアップデートするため、マッパーの PushData に対する干渉は一切生じません。全体プロセスを下図に示します。

Sort On Read

ランダム読み取りの問題を回避するため、RSS は Sort On Read の戦略を採用します。具体的には、ファイルスプリットの最初の範囲読み取りがソートをトリガーし (範囲外の読み取りはトリガーしません)、ソートされたファイルはその位置インデックスとともにディスクに書き戻されます。後続の範囲読み取りはシーケンシャル読み取りを保証します。下図に示します。

複数のサブリデューサーが同じファイルスプリットのソートを待機する問題を回避するため、各サブリデューサーがスプリットを読み取る順序を分散させます。下図に示します。

ソートの最適化

Sort On Read は冗長なランダム読み取りを効果的に回避できますが、ファイルスプリット (256 MB) のソートが必要です。本セクションではソートの実装とコストについて議論します。ファイルソートはファイル読み取り、MapId のソート、ファイル書き込みの 3 つのステップで構成されます。RSS のデフォルトブロックサイズは 256 KB で、ブロック数は約 1,000 です。したがって、ソートプロセス自体は非常に高速で、主なオーバーヘッドはファイルの読み書きです。ソートプロセス全体にはおおまかに 3 つの方式があります。

1. ファイルサイズ分のメモリを事前割り当てし、ファイル全体を一括で読み取り、MapId を解析してソートし、ブロックを MapId の順序でディスクに書き戻す。

2. メモリを割り当てず、各ブロックの位置にシークして MapId を解析してソートし、元ファイルのブロックを MapId の順序で新ファイルに転送する。

3. 小メモリブロック (例：256 KB) を割り当て、ファイル全体を順番に読み取り、MapId を解析してソートし、元ファイルのブロックを MapId の順序で新ファイルに転送する。

IO の観点から見ると、一見すると方式 1 は十分なメモリを使用してシーケンシャル読み書きを実現しています。方式 2 はランダム読み取りとランダム書き込みがあります。方式 3 はランダム書き込みがあります。直感的には方式 1 のパフォーマンスが優れているように見えます。しかし、PageCache の存在により、方式 3 では書き込み時に元のファイルのほぼ全体が PageCache にキャッシュされている状態になるため、方式 3 のパフォーマンスが優れています。下図に示します。

同時に、方式 3 はプロセスの追加メモリを占有する必要がないため、RSS は方式 3 のアルゴリズムを採用しています。また、Sort On Read とソートなしのランダム読み取り方式およびインデックスのみの方式との比較もテストしました。下図に示します。

全体プロセス

RSS のスキュー結合最適化サポートの全体プロセスを下図に示します。

RSS のフロー制御

フロー制御の主な目的は、RSS ワーカーメモリの溢れ出しを防止することです。フロー制御には通常 2 つの方法があります。

1. クライアントが各 PushData の前にワーカーに対してメモリを予約し、予約が成功した後にのみプッシュをトリガーする。

2. ワーカー側でのバックプレッシャー。

PushData は非常に高頻度でパフォーマンスクリティカルな操作であるため、プッシュごとに追加の RPC 対話を行うコストが高すぎるため、バックプレッシャーストラテジーを採用しました。ワーカーの観点から見ると、流入データには 2 つのソースがあります。

1. クライアントからプッシュされるデータ

2. マスターコピーから送信されるデータ

下図に示すように、Worker2 はマッパーからプッシュされた Partition3 のデータと、Worker1 から送信された Partition1 のレプリカデータの両方を受信し、Partition3 のデータを対応するスレーブレプリカに送信します。

このうち、マッパーからのデータは以下の条件がすべて同時に満たされた場合にのみメモリが解放されます。

1. レプリケーションの実行が成功

2. データの書き込みが成功

プライマリレプリカからプッシュされたデータは以下の条件が満たされた場合にのみメモリが解放されます。

1. データの書き込みが成功

フロー制御ストラテジーを設計する際は、流入制限 (データの流入を減らす) だけでなく、排出 (メモリがタイムリーに解放されること) も考慮する必要があります。具体的には、高水位用に 2 つのメモリしきい値 (それぞれメモリ使用量 85% と 95% に対応)、低水位用に 1 つのメモリしきい値 (メモリ使用量 50%) を定義します。高水位の第 1 しきい値に達すると、フロー制御がトリガーされ、マッパーからプッシュされるデータが一時停止され、同時にディスクへの強制フラッシュが実行され、排出の目標を達成します。マッパーからの流入のみを制限しても、プライマリレプリカからのフローは制御できません。そのため、高水位の第 2 レベルを定義します。このしきい値に達すると、プライマリレプリカから送信されるデータも同時に一時停止されます。水位が低水位を下回ると、通常の状態に戻ります。全体プロセスを下図に示します。

パフォーマンステスト

RSS とネイティブの External Shuffle Service (ESS) のパフォーマンスを比較するため、Spark 3.2.0 で AQE を有効にしてテストしました。RSS は混在モードを採用しており、追加のマシンリソースを占有しません。また、RSS が使用するメモリは 8 GB で、マシンメモリ (352 GB) のわずか 2.3% です。具体的な環境は以下のとおりです。

テスト環境

ハードウェア：

ヘッダーマシングループ 1 x ecs.g5.4xlarge

ワーカーマシングループ 8 x ecs.d2c.24xlarge、96 CPU、352 GB、12 x 3700 GB HDD。

Spark AQE 関連設定：

spark.sql.adaptive.enabled true

spark.sql.adaptive.coalescePartitions.enabled true

spark.sql.adaptive.coalescePartitions.initialPartitionNum 1000

spark.sql.adaptive.skewJoin.enabled true

spark.sql.adaptive.localShuffleReader.enabled false

RSS 関連設定：

RSS_MASTER_MEMORY=2g

RSS_WORKER_MEMORY=1g

RSS_WORKER_OFFHEAP_MEMORY=7g

TPC-DS 10 TB テストセット

10 TB TPC-DS をテストした結果、エンドツーエンドで ESS は 11,734 秒、RSS シングルレプリカ / 2 レプリカはそれぞれ 8,971 秒 / 10,110 秒で、ESS より 23.5% / 13.8% 高速化されました。下図に示します。RSS で 2 レプリカを有効にした際にネットワーク帯域幅が上限に達することが確認されており、これが 2 レプリカがシングルレプリカより遅い主な要因です。