アダプティブクエリ実行：実行時にSpark SQLを高速化

「データレイクハウスの台頭」を読むことで、データウェアハウスの父であるビル・インモンと共に、なぜレイクハウスが未来のデータアーキテクチャなのかを探求しましょう。

これは、Databricks Apache Spark エンジニアリングチーム（Wenchen Fan、Herman van Hovell、MaryAnn Xue）と Intel エンジニアリングチーム（Ke Jia、Haifeng Chen、Carson Wang）による共同エンジニアリングの成果です。

ここで説明するソリューションをデモするには、AQEノートブックを参照してください。または、Databricks Lakehouse Platform の内部構造をさらに深く掘り下げてみましょう。

長年にわたり、Spark SQL のクエリオプティマイザーとプランナーを改善し、高品質なクエリ実行プランを生成するための、広範かつ継続的な取り組みが行われてきました。最も大きな改善の1つは、コストベースの最適化フレームワークであり、これは行数、一意の値の数、NULL値、最大/最小値などのさまざまなデータ統計を収集して活用し、Spark がより良いプランを選択できるようにします。これらのコストベースの最適化手法の例としては、適切な結合タイプ（ブロードキャストハッシュ結合 vs ソートマージ結合）の選択、ハッシュ結合における正しいビルドサイドの選択、または複数結合における結合順序の調整などが挙げられます。しかし、古い統計情報や不完全なカーディナリティ推定は、最適ではないクエリプランにつながる可能性があります。Adaptive Query Execution は、Apache Spark^TM 3.0 の今後のリリースで新しくなり、Databricks Runtime 7.0 で利用可能になりました。これは、クエリ実行プロセス中に収集された実行時統計に基づいてクエリプランを再最適化および調整することで、このような問題に対処しようとしています。

Adaptive Query Execution (AQE) フレームワーク

Adaptive Query Execution における最も重要な質問の1つは、いつ再最適化するかということです。Spark オペレータはしばしばパイプライン化され、並列プロセスで実行されます。しかし、シャッフルまたはブロードキャスト交換は、このパイプラインを中断します。これらをマテリアライゼーションポイントと呼び、「クエリステージ」という用語を使用して、クエリ内のこれらのマテリアライゼーションポイントによって区切られたサブセクションを示します。各クエリステージは中間結果を具体化し、次のステージは、マテリアライゼーションを実行しているすべての並列プロセスが完了した場合にのみ進行できます。これは再最適化の自然な機会を提供します。なぜなら、すべてのパーティションのデータ統計が利用可能になり、後続の操作がまだ開始されていないからです。

クエリが開始されると、Adaptive Query Execution フレームワークは最初にすべてのリーフステージ（他のステージに依存しないステージ）を開始します。これらのステージの1つ以上がマテリアライゼーションを完了するとすぐに、フレームワークは物理クエリプランでそれらを完了としてマークし、完了したステージから取得した実行時統計を使用して論理クエリプランを更新します。これらの新しい統計に基づいて、フレームワークはオプティマイザー（選択された論理最適化ルールのリストを含む）、物理プランナー、およびパーティションの結合、スキュー結合の処理などの通常の物理ルールとアダプティブ実行固有のルールを含む物理最適化ルールを実行します。これで、完了したステージを持つ新しく最適化されたクエリプランが得られたので、アダプティブ実行フレームワークは、子ステージがすべて具体化された新しいクエリステージを検索して実行し、上記の実行-再最適化-実行プロセスをクエリ全体が完了するまで繰り返します。

Spark 3.0 では、AQE フレームワークは3つの機能とともに提供されます。

• シャッフルパーティションの動的結合
• 結合戦略の動的切り替え
• スキュー結合の動的最適化

次のセクションでは、これらの3つの機能について詳しく説明します。

シャッフルパーティションの動的結合

Spark で非常に大きなデータを処理するクエリを実行する場合、シャッフルは、他の多くのことの中でも、クエリパフォーマンスに非常に重要な影響を与えます。シャッフルは高価なオペレータであり、データをネットワーク経由で移動する必要があるため、下流オペレータが必要とするようにデータが再分散されます。

シャッフルの重要な特性の1つはパーティションの数です。最適なパーティション数はデータに依存しますが、データサイズはステージごと、クエリごとに大きく異なる可能性があるため、この数を調整するのは困難です。

1. パーティションが少なすぎると、各パーティションのデータサイズが非常に大きくなる可能性があり、これらの大きなパーティションを処理するタスクはディスクにデータをスピルする必要があるかもしれません（例：ソートまたは集計が関与する場合）。その結果、クエリが遅くなります。
2. パーティションが多すぎると、各パーティションのデータサイズが非常に小さくなる可能性があり、シャッフルブロックを読み取るために多くの小さなネットワークデータフェッチが発生します。これは、非効率的な I/O パターンによりクエリを遅くする可能性もあります。多数のタスクを持つことは、Spark タスクスケジューラにも負担をかけます。

この問題を解決するために、最初に比較的多くのシャッフルパーティションを設定し、次にシャッフルファイル統計を確認して、実行時に隣接する小さなパーティションをより大きなパーティションに結合することができます。

たとえば、SELECT max(i) FROM tbl GROUP BY j というクエリを実行しているとします。入力データ tbl は非常に小さいため、グループ化前はパーティションは2つしかありません。初期シャッフルパーティション数は5に設定されているため、ローカルグループ化の後、部分的にグループ化されたデータは5つのパーティションにシャッフルされます。AQE がない場合、Spark は最終集計を実行するために5つのタスクを開始します。しかし、ここには3つの非常に小さなパーティションがあり、それぞれに個別のタスクを開始するのは無駄です。

代わりに、AQE はこれらの3つの小さなパーティションを1つに結合し、その結果、最終集計は5つではなく3つのタスクのみを実行する必要があります。

結合戦略の動的切り替え

Spark は多くの結合戦略をサポートしていますが、その中でもブロードキャストハッシュ結合は、結合の一方のサイドがメモリにうまく収まる場合、通常は最もパフォーマンスが高くなります。このため、結合関係の推定サイズがブロードキャストサイズしきい値を下回る場合、Spark はブロードキャストハッシュ結合を計画します。しかし、非常に選択性の高いフィルタの存在や、結合関係が単なるスキャンではなく一連の複雑なオペレータであるなど、このサイズ推定を誤らせる可能性のある多くのことがあります。

この問題を解決するために、AQE は最も正確な結合関係サイズに基づいて、実行時に結合戦略を再計画します。次の例で見られるように、結合の右側は推定よりもはるかに小さいことが判明し、ブロードキャストできるほど小さいことも判明したため、AQE の再最適化後、静的に計画されたソートマージ結合はブロードキャストハッシュ結合に変換されます。

実行時に変換されたブロードキャストハッシュ結合の場合、通常のシャッフルをローカライズされたシャッフル（つまり、リデューサーごとではなくマッパーごとに読み取るシャッフル）にさらに最適化して、ネットワークトラフィックを削減できます。

スキュー結合の動的最適化

データスキューは、クラスター内のパーティション間でデータが不均一に分散している場合に発生します。深刻なスキューは、特に結合において、クエリパフォーマンスを大幅に低下させる可能性があります。AQE スキュー結合最適化は、シャッフルファイル統計からこのようなスキューを自動的に検出します。次に、スキューしたパーティションをより小さなサブパーティションに分割し、それらをそれぞれ相手側の対応するパーティションと結合します。

テーブル A とテーブル B の結合の例を見てみましょう。テーブル A には、他のパーティションよりも大幅に大きいパーティション A0 があります。

この最適化がない場合、4つのタスクがソートマージ結合を実行し、1つのタスクがはるかに長い時間を要していました。この最適化の後、結合を実行するタスクは5つになりますが、各タスクにかかる時間はほぼ同じになり、全体的なパフォーマンスが向上します。

TPC-DSパフォーマンス向上におけるAQEの効果

TPC-DSデータとクエリを使用した実験では、Adaptive Query Executionによりクエリパフォーマンスが最大8倍向上し、32のクエリで1.1倍以上の速度向上が見られました。以下は、AQEによって最もパフォーマンスが向上した10個のTPC-DSクエリのチャートです。

これらの改善のほとんどは、動的パーティション結合と動的結合戦略の切り替えによるものです。TPC-DSデータはランダムに生成されるためスキューが発生しませんが、本番環境のワークロードではAQEの3つの機能すべてが活用され、さらに大きな改善が見られました。

AQEの有効化

AQEは、SQL設定spark.sql.adaptive.enabledをtrue（Spark 3.0ではデフォルトでfalse）に設定することで有効化できます。これは、クエリが以下の基準を満たす場合に適用されます。

• ストリーミングクエリではない
• 少なくとも1つの交換（通常は結合、集計、またはウィンドウ演算子がある場合）または1つのサブクエリが含まれている

クエリ最適化を静的統計情報への依存度を下げることで、AQEはSparkのコストベース最適化における最大の課題の1つ、つまり統計情報収集のオーバーヘッドと推定精度のバランスの問題を解決しました。最高の推定精度と計画結果を達成するには、通常、詳細で最新の統計情報（列ヒストグラムなど）を維持する必要があります。これらの統計情報は、選択率とカーディナリティの推定を改善したり、データスキューを検出したりするために使用できますが、収集にはコストがかかる場合があります。AQEは、これらの統計情報や手動チューニングの必要性を大幅に排除しました。さらに、AQEは、任意のUDFや予測不可能なデータセットの変更（データサイズの突然の増加または減少、頻繁でランダムなデータスキューなど）が存在する場合でも、SQLクエリ最適化をより回復力のあるものにしました。事前にデータを「知る」必要はもうありません。AQEはクエリの実行中にデータを把握し、クエリプランを改善して、分析とシステムパフォーマンスを向上させます。

Spark 3.0の詳細については、プレビューウェビナーをご覧ください。Spark 3.0のAQEを、Databricks Runtime 7.0の一部としてDatabricksで無料でお試しください。

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事