:Spark SQL のアダプティブ実行

現在、Spark SQL の削減ステージのタスク数は spark.sql.shuffle.partition パラメーター (デフォルト値は 200) の値によって異なります。 ジョブにこのパラメーターが指定されると、すべてのステージにおける削減タスクの数は、ジョブが実行中のときと同じ値になります。

ジョブごとに、また 1 つのジョブの削減ステージごとに、実際のデータサイズはかなり異なる場合があります。 たとえば、削減ステージで処理されるデータのサイズは 10 MB の場合もあれば 100 GB の場合もあります。 同じ値を使用してパラメーターを指定すると、実際の処理効率に大きな影響を与えます。 たとえば、10 MB のデータは 1 つのタスクだけで処理可能です。 spark.sql.shuffle.partition の値をデフォルト値の 200 に設定すると、10 MB のデータは 200 個のタスクによる処理用に分割されます。 これにより、スケジューリングのオーバーヘッドが増加し、処理効率が低下します。

Spark SQL のアダプティブ実行フレームワークは、シャッフルパーティション番号の範囲を設定することで、異なるジョブの異なるステージに合わせて範囲内の削減タスクの数を動的に調整できます。

これにより、最適化にかかるコストが大幅に削減されます (固定値を決める必要がありません)。 さらに、1 つのジョブのさまざまなステージにおける削減タスクの数を動的に調整できます。

データスキューは、SQL 結合操作でよく見られる問題です。 特定のタスクが処理に巻き込むデータが多すぎるというシナリオで、ロングテールにつながります。 現在、Spark SQL は、データの歪みを最適化しません。

Spark SQL の Adaptive Execution フレームワークは、歪んだデータを自動的に検出し、実行時に最適化を実行します。

SparkSQL は、パーティション内の歪んだデータを分割し、複数のタスクを介してデータを処理してから、SQL 結合操作を介して結果を結合することにより、データの歪みを最適化します。

Spark SQL の Catalyst オプティマイザは、SQL 文から物理実行プランに変換される論理プランを変換し、それらの物理実行プランを実行します。 ただし、Catalyst が作成する物理実行プランは、統計の欠如または不正確さが原因で最適ではない場合があります。 たとえば、Spark SQL は BroadcastJoin ではなく SortMergeJoinExec を選択することはできますが、BroadcastJoin はシナリオで最適なオプションです。

Spark SQL の Adaptive Execution フレームワークは、シャッフルステージでのシャッフル書き込みのサイズをベースにクエリパフォーマンスを改善するため、SortMergeJoin ではなく BroadcastJoin を使用するかどうか決定します。