パフォーマンス チューニング

• メモリへのデータのキャッシュ
• 他の設定オプション
• SQL クエリのための Join ストラテジ ヒント
• SQL クエリのための Coalesce ヒント
• 適応クエリ実行
  • シャッフル後のパーティションの結合
  • sort-merge join から broadcast join への変換
  • ソートマージjoinからシャッフルハッシュjoinへの変換
  • スキューJoinの最適化

ちょっとした次善策として、メモリにデータをキャッシュ、あるいは幾つかの実験的なオプションを調整することでパフォーマンスを改善することができます。

メモリへのデータのキャッシュ

Spark SQL は spark.catalog.cacheTable("tableName") あるいはdataFrame.cache()を呼ぶことでインメモリのコラム形式のフォーマットを使ってテーブルをキャッシュすることができます。そして、Spark SQLは必要なカラムだけをスキャンし、メモリの使用量とGCの圧力を最小化するために圧縮を自動的に調整するでしょう。メモリからテーブルを削除するためにspark.catalog.uncacheTable("tableName")またはdataFrame.unpersist()を呼ぶことができます。

メモリ内キャッシングの設定はSparkSessionのsetConfメソッドあるいは SQLを使ってSET key=valueコマンドを実行することで行うことができます。

他の設定オプション

以下のオプションもクエリ実行のパフォーマンスを調整するために使用することができます。これらのオプションはもっと多くの最適化が自動的に行われるため、将来のリリースでは非推奨になるかもしれません。

SQL クエリのための Join ストラテジ ヒント

joinの戦略のヒント、つまりBROADCAST, MERGE, SHUFFLE_HASH, SHUFFLE_REPLICATE_NL は、別のリレーションと結合する時に指定されたリレーションでヒント付きの戦略を使うようにSparkに指示します。For example, when the BROADCAST hint is used on table ‘t1’, broadcast join (either broadcast hash join or broadcast nested loop join depending on whether there is any equi-join key) with ‘t1’ as the build side will be prioritized by Spark even if the size of table ‘t1’ suggested by the statistics is above the configuration spark.sql.autoBroadcastJoinThreshold.

joinの両側で異なるjoin戦略ヒントが指定された場合、Sparkは以下の順で優先します。BROADCAST、MERGE、SHUFFLE_HASH、SHUFFLE_REPLICATE_NL。両側がBROADCASTヒントあるいはSHUFFLE_HASHヒントで指定されている場合、Sparkはjoin型とリレーションのサイズに基づいてビルド側を選択します。

特定の戦略が全てのjoin型をサポートするわけではないため、Sparkがヒントで指定されたjoin戦略を選択する保証はないことに注意してください。

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show()

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show();

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show()

src <- sql("SELECT * FROM src")
records <- sql("SELECT * FROM records")
head(join(src, hint(records, "broadcast"), src$key == records$key))

-- We accept BROADCAST, BROADCASTJOIN and MAPJOIN for broadcast hint
SELECT /*+ BROADCAST(r) */ * FROM records r JOIN src s ON r.key = s.key

詳細については、Join ヒントのドキュメントを参照してください。

SQL クエリのための Coalesce ヒント

Coalesceヒントにより、Spark SQL ユーザはDataset APIのcoalesce, repartition, repartitionByRangeと同じように、出力ファイルの数を制御できます。パフォーマンスの調整と出力ファイルの削減に使われます。“COALESCE”ヒントはパラメータとしてパーティション番号のみを取ります。“REPARTITION”ヒントには、パラメータとして、パーティション番号、列、またはそれらの両方/どちらかを取ります。“REPARTITION_BY_RANGE” ヒントは列名が必須で、パーティション番号はオプションです。

SELECT /*+ COALESCE(3) */ * FROM t
SELECT /*+ REPARTITION(3) */ * FROM t
SELECT /*+ REPARTITION(c) */ * FROM t
SELECT /*+ REPARTITION(3, c) */ * FROM t
SELECT /*+ REPARTITION */ * FROM t
SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t
SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t
SELECT /*+ REBALANCE */ * FROM t
SELECT /*+ REBALANCE(c) */ * FROM t

詳細については、パーティショニング ヒントのドキュメントを参照してください。

適応クエリ実行

適応クエリ実行(AQE)は、ランタイム統計を利用して最も効率的なクエリ実行プランを選択するSpark SQLの最適化手法で、Apache Spark 3.2.0からデフォルトで有効になっています。Spark SQLはアンブレラ構成としてspark.sql.adaptive.enabledによって、AOEのオンとオフを切り替えることができます。Spark 3.0の時点で、AOEには3つの主要な機能があります: シャッフル後のパーティションの結合、ソートマージjoinからブロードキャストjoinへの返還、スキューjoinの最適化です。

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

この機能は、spark.sql.adaptive.enabledとspark.sql.adaptive.coalescePartitions.enabled設定の両方がtrueの時に、マップ出力統計に基づいてシャッフル後のパーティションを結合します。この機能により、クエリを実行する時のシャッフルパーティション番号の調整が簡単になります。データセットに合わせて適切なシャッフルパーティション番号を設定する必要はありません。spark.sql.adaptive.coalescePartitions.initialPartitionNum設定でシャッフルパーティションの初期数を十分大きく設定すると、Sparkは実行時に適切なシャッフルパーティション番号を選択できます。

sort-merge join から broadcast join への変換

AQEは、join側の実行時統計が適応型ブロードキャストハッシュjoinの閾値よりも小さい場合に、ソートマージjoinをブロードキャストハッシュjoinに変換します。これは最初にブロードキャストハッシュjoinを計画するほどには効率的ではありませんが、両方のjoin側のソートを保存し、シャッフルファイルをローカルで読み込んでネットワークトラフィックを節約できるため、ソートマージjoinを実行し続けるよりも優れています(spark.sql.adaptive.localShuffleReader.enabled がtrueの場合)

ソートマージjoinからシャッフルハッシュjoinへの変換

AQE converts sort-merge join to shuffled hash join when all post shuffle partitions are smaller than a threshold, the max threshold can see the config spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold.

スキューJoinの最適化

データスキューはjoinクエリのパフォーマンスを大幅に低下させる可能性があります。この機能は、スキューされたタスクをほぼ均等なサイズのタスクに分割(および必要に応じて複製)することで、ソートマージjoinのスキューを動的に処理します。spark.sql.adaptive.enabledとspark.sql.adaptive.skewJoin.enabled設定の両方が有効になっている場合に効果があります。