SQLを高速に : DatabricksにおけるAdaptive Query Execution

[Faster SQL: Adaptive Query Execution in Databricks \- The Databricks Blog](https://databricks.com/blog/2020/10/21/faster-sql-adaptive-query-execution-in-databricks.html)の翻訳です。

:::note warn
2020年の記事です。
:::

今年の初めに、Databricksでは、Spark 3.0およびDatabricks 7.0で提供された、[全く新しいAdaptive Query Execution](https://qiita.com/taka_yayoi/items/f01d1bbbaeda2857d5f7)に関する記事を投稿しました。この記事は技術者からの多大なる興味と議論を引き起こしました。本日、Databricksランタイム、DBR 7.3において、デフォルトでAdaptive Query Executionが有効化されたことを嬉しく思っています。

AQEは、不十分、不正確、鮮度の低いオプティマイザの統計情報によって引き起こされるクエリー実行計画の非効率性、柔軟性の欠如への対策を狙いとした、実行時SQL最適化フレームワークです。我々はAQEの機能拡張を続けており、現状皆様が最もAQEが効率的と考えるであろう特定のユースケースとなります。

# シャッフルの最適化

クエリー性能においてSparkのシャッフルが重要な部分となりますが、適切な*シャッフルパーティションの数*を選ぶことは、Sparkユーザーにとって常に大きな問題となっています。クエリーごとにデータサイズが異なり、1つのクエリーにおいてもステージ間でサイズが異なるため、同じ*シャッフルパーティション数*を使うと、Sparkスケジューラーの利用が非効率的になる小規模のタスクや、ディスクのspillや過大なガーベージコレクション(GC)を引き起こす大規模なタスクを引き起こします。

今では、AQEはクエリーのそれぞれのステージで、map側のシャッフルアウトプットのサイズに基づいて*シャッフルパーティション数*を自動で調整します。ステージごとにデータサイズは増減しますが、タスクのサイズは大きすぎず、小さすぎず概ね同じものになります。

しかし、AQEはmap側のパーティション数を自動で設定しないことに注意する必要があります。このことは、AQEの機能が完全に動作するようにするためには、SQL設定`spark.sql.shuffle.partitions`を最初は比較的大きい値に設定することをお勧めします。あるいは、代替案として、`spark.databricks.adaptive.autoOptimizeShuffle.enabled`を`true`に設定することで、Databricksの最新機能「シャッフルの自動最適化」を有効化することもできます。

# Join戦略の選択

Sparkのオプティマイザにおける最も重要なコストベースの意思決定は、joinのリレーションの推定サイズに基づくjoin戦略の選択です。しかし、この推定は両方の方向で間違うケースがあります。過度の見積もりによる非効率的なjoin戦略、さらに悪いことには、見積もり不足によるアウトオブメモリーエラーになる場合があります。

AQEは、実行時により高速なbroadcast hash joinに切り替えることでトラブルのないソリューションを提供します。

# Skew Joinの取り扱い

データの偏り(skew)は、データが均一に分配されておらず、特にsort merge joinにおいて、ボトルネックと深刻なパフォーマンスの悪化を引き起こす一般的な問題です。長時間実行されるタスクは落伍者となり、ステージ全体をスローダウンさせます。さらに、これらの偏ったパーティションにおいてメモリーからディスクにデータが溢れることで、スローダウンの影響を悪化させます。

データの偏りの予測不可能な性質は、クエリーのヒントを用いたとしても、静的なオプティマイザが自動で偏りを取り扱うことを困難にします。実行時の統計情報を収集することで、AQEは実行時にskew joinを検知し、偏りのあるパーティションを小規模なサブパーティション分割することで、クエリー性能におけるネガティブなインパクトを排除します。

# AQEクエリープランを理解する

AQEクエリープランにおける主要な違いは、多くの場合で実行が進むにつれてプランが進化するというものです。実行に関する詳細情報を提供するために、いくつかのAQE固有のプランノードが導入されました。さらに、AQEは、初期クエリープラン、最終的クエリープランの両方で表示される新たなクエリープランの文字列フォーマットを使用します。このセクションでは、ユーザーが新たなAQEクエリープランになれるようにするお手伝いをし、クエリーにおけるAQEの効果をどのように特定するのかを説明します。

## AdaptiveSparkPlanノード

AQEが適用されたクエリーには、それぞれのクエリー、あるいはサブクエリーのルートノードとして、1つ以上のAdaptiveSparkPlanノードが存在します。実行前、実行中には、`isFinalPlan`フラグは`false`となります。クエリーが完了すると、このフラグは`true`に変わり、AdaptiveSparkPlanノードの下のプランはもう変化しません。
![](https://databricks.com/wp-content/uploads/2020/10/blog-adaptive-query-1.png)

## CustomShuffleReaderノード

CustomShuffleReaderノードはAQEにおける最適化の鍵となります。これは、シャッフルのmapステージで収集された統計情報に基づいて、シャッフル後のパーティション数を動的に調整します。Spark UIにおいては、ユーザーはシャッフルされたパーティションに適用された最適化を確認するために、ノード上にカーソルをホバーすることができます。

CustomShuffleReaderのフラグが`coalesced`になっている場合、AQEはターゲットのパーティションサイズに基づいて小規模のパーティションを検知し、シャッフル後に結合したことを意味します。このノードの詳細では、シャッフルパーティションの数と結合後のパーティションサイズを確認することができます。
![](https://databricks.com/wp-content/uploads/2020/10/blog-adaptive-query-2.png)

CustomShuffleReaderのフラグが`skewed`になっている場合、AQEがsort-merge joinオペレーションの前に、1つ以上のパーティションに偏りがあることを検知したこを意味します。このノードの詳細では、偏りのあるパーィションの数と、偏りのあるパーティションから分割された新たなパーティションの数が表示されます。
![](https://databricks.com/wp-content/uploads/2020/10/blog-adaptive-query-3.png)

同時に2つの効果が得られる場合もあります。
![](https://databricks.com/wp-content/uploads/2020/10/blog-adaptive-query-4.png)

## Join戦略変更の検知

AQEの最適化前後におけるクエリープランのjoinノードの変化を比較することで、join戦略の変更を特定することができます。DBR 7.3では、AQEクエリープランの文字列には、初期プラン(AQE最適化の適用前のプラン)と現在のプラン、最終的なプランが含まれます。これにより、クエリーに適用されたAQEの最適化を、より詳細に確認できるようになります。以下には、sort-merge joinからbroadcast-hash joinに変更された新規クエリープランの礼を示します。
![](https://databricks.com/wp-content/uploads/2020/10/blog-adaptive-query-5.png)

Spark UIでは、現在のプランのみが表示されます。Spark UIで効果を確認するためには、ユーザーはクエリー実行前と実行後でプランの図を比較する必要があります。
![](https://databricks.com/wp-content/uploads/2020/10/blog-adaptive-query-6.png)

## Skew Joinの検知

Skew join最適化の効果は、joinノード名で識別することができます。

Spark UI
![](https://databricks.com/wp-content/uploads/2020/10/blog-adaptive-query-7.png)

クエリープランの文字列
![](https://docs.databricks.com/_images/skew-join-string.png)

Adaptive Query Executionは、クエリー実行をより効率的なものにするために実行時の統計情報を活用します。他の最適化技術と異なり、最適なシャッフル後のパーティションサイズ、パーティション数を決定し、join戦略を切り替え、skew joinをうまく取り扱います。AQEの詳細については、Spark + AIサミットの[Adaptive Query Execution: Speeding Up Spark SQL at Runtime](https://databricks.com/session_na20/adaptive-query-execution-speeding-up-spark-sql-at-runtime)、および[AQEユーザーガイド](https://docs.databricks.com/spark/latest/spark-sql/aqe.html)をご覧ください。Databricksランタイム7.3以降でAQEの新機能を体験してみてください。

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)