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

Spark は、DataFrame または SQL ワークロードのパフォーマンスをチューニングするための多くのテクニックを提供しています。これらのテクニックは、広義には、データのキャッシュ、データセットのパーティショニング方法の変更、最適な結合戦略の選択、およびオプティマイザがより効率的な実行計画を構築するために使用できる追加情報を提供することを含みます。

• データのキャッシュ
• パーティションのチューニング
  • Coalesce ヒント
• 統計情報の活用
• 結合戦略の最適化
  • 結合の自動ブロードキャスト
  • 結合戦略ヒント
• アダプティブクエリ実行
  • シャッフル後パーティションの結合
  • スキューズしたシャッフルパーティションの分割
  • ソートマージ結合をブロードキャスト結合に変換
  • ソートマージ結合をシャッフルハッシュ結合に変換
  • スキューズ結合の最適化
  • 高度なカスタマイズ
• ストレージパーティション結合

データのキャッシュ

Spark SQL は、spark.catalog.cacheTable("tableName") または dataFrame.cache() を呼び出すことで、インメモリ列フォーマットを使用してテーブルをキャッシュできます。その後、Spark SQL は必要な列のみをスキャンし、メモリ使用量と GC 負荷を最小限に抑えるために圧縮を自動的にチューニングします。spark.catalog.uncacheTable("tableName") または dataFrame.unpersist() を呼び出すことで、メモリからテーブルを削除できます。

インメモリキャッシングの設定は、spark.conf.set を使用するか、SQL を使用して SET key=value コマンドを実行することで行うことができます。

パーティションのチューニング

Coalesce ヒント

Coalesce ヒントは、Spark SQL ユーザーが Dataset API の `coalesce`、`repartition`、`repartitionByRange` と同様に出力ファイルを制御できるようにします。これらは、パフォーマンスチューニングや出力ファイルの削減に使用できます。「COALESCE」ヒントはパーティション数のみをパラメータとして持ちます。「REPARTITION」ヒントは、パーティション数、列、またはその両方/なしをパラメータとして持ちます。「REPARTITION_BY_RANGE」ヒントは列名を必須とし、パーティション数はオプションです。「REBALANCE」ヒントは、初期パーティション数、列、またはその両方/なしをパラメータとして持ちます。

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(3) */ * FROM t;
SELECT /*+ REBALANCE(c) */ * FROM t;
SELECT /*+ REBALANCE(3, c) */ * FROM t;

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

統計情報の活用

Apache Spark が多くの可能なオプションの中から最適な実行計画を選択する能力は、部分的には実行計画の各ノード（読み取り、フィルタリング、結合など）から出力される行数を推定することによって決定されます。これらの推定は、Spark がいくつかの方法で利用可能になった統計情報に基づいています。

• データソース: Parquet ファイルのメタデータ内のカウントや最小/最大値など、Spark が基になるデータソースから直接読み取る統計情報。これらの統計情報は、基になるデータソースによって維持されます。
• カタログ: Hive Metastore のような、Spark がカタログから読み取る統計情報。これらの統計情報は、ANALYZE TABLE を実行するたびに収集または更新されます。
• 実行時: クエリの実行中に Spark が自身で計算する統計情報。これは、アダプティブクエリ実行フレームワーク の一部です。

統計情報が不足していたり不正確だったりすると、Spark が最適な計画を選択する能力が妨げられ、クエリのパフォーマンスが低下する可能性があります。そのため、Spark で利用可能な統計情報と、クエリの計画および実行中に Spark が行う推定を調べることは役立ちます。

• データオブジェクト統計情報: DESCRIBE EXTENDED を使用して、テーブルまたは列の統計情報を検査できます。
• クエリプラン推定値: EXPLAIN COST または DataFrame.explain(mode="cost") を介して、最適化されたクエリプランで Spark のコスト推定値を検査できます。
• 実行時統計情報: クエリの実行中に、SQL UI の「詳細」セクションでこれらの統計情報を検査できます。実行中のプランで Statistics(..., isRuntime=true) を探してください。

結合戦略の最適化

結合の自動ブロードキャスト

結合戦略ヒント

結合戦略ヒント、つまり BROADCAST、MERGE、SHUFFLE_HASH、SHUFFLE_REPLICATE_NL は、Spark に、指定された各リレーションで、別 のリレーションと結合する際にヒントされた戦略を使用するように指示します。たとえば、テーブル 't1' で BROADCAST ヒントが使用されている場合、統計情報によって示されるテーブル 't1' のサイズが `spark.sql.autoBroadcastJoinThreshold` 設定を超えている場合でも、't1' をビルドサイドとするブロードキャスト結合（ブロードキャストハッシュ結合またはブロードキャストネストドループ結合（等価結合キーの有無による））が Spark によって優先されます。

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

特定の戦略がすべての結合タイプをサポートしているとは限らないため、Spark がヒントで指定された結合戦略を選択するという保証はありません。

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

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

アダプティブクエリ実行

アダプティブクエリ実行（AQE）は、Spark SQL の最適化テクニックであり、実行時の統計情報を使用して最も効率的なクエリ実行計画を選択します。これは Apache Spark 3.2.0 以降、デフォルトで有効になっています。Spark SQL は、spark.sql.adaptive.enabled を包括的な設定として使用して、AQE をオンまたはオフにできます。

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

この機能は、`spark.sql.adaptive.enabled` および `spark.sql.adaptive.coalescePartitions.enabled` の両方の設定が true の場合に、マップ出力統計情報に基づいて、シャッフル後パーティションを結合します。この機能により、クエリの実行時にシャッフルパーティション数をチューニングする手間が省けます。データセットに適合する適切なシャッフルパーティション数を設定する必要はありません。`spark.sql.adaptive.coalescePartitions.initialPartitionNum` 設定で十分な数の初期シャッフルパーティションを設定すれば、Spark は実行時に適切なシャッフルパーティション数を選択できます。

スキューズしたシャッフルパーティションの分割

ソートマージ結合をブロードキャスト結合に変換

AQE は、実行時統計情報がアダプティブブロードキャストハッシュ結合のしきい値より小さい場合、ソートマージ結合をブロードキャストハッシュ結合に変換します。これは、最初にブロードキャストハッシュ結合を計画するほど効率的ではありませんが、ソートマージ結合を続行するよりは優れています。ソートマージ結合を続行するより優れているのは、両方の結合サイドのソートを回避し、ローカルでシャッフルファイルを読み取ってネットワークトラフィックを節約できるためです（`spark.sql.adaptive.localShuffleReader.enabled` が true の場合）。

ソートマージ結合をシャッフルハッシュ結合に変換

AQE は、すべてのシャッフル後パーティションが `spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold` で構成されたしきい値よりも小さい場合、ソートマージ結合をシャッフルハッシュ結合に変換します。

スキューズ結合の最適化

データスキューズは、結合クエリのパフォーマンスを著しく低下させる可能性があります。この機能は、スキューズしたタスクをほぼ均等なサイズのタスクに分割する（必要に応じてレプリケートする）ことによって、ソートマージ結合におけるスキューズを動的に処理します。これは、`spark.sql.adaptive.enabled` および `spark.sql.adaptive.skewJoin.enabled` の両方の設定が有効な場合に有効になります。

高度なカスタマイズ

カスタムコスト評価クラスを提供するか、AQE オプティマイザールールを除外することで、AQE の動作の詳細を制御できます。

ストレージパーティション結合

ストレージパーティション結合（SPJ）は、Spark SQL の最適化テクニックであり、既存のストレージレイアウトを利用してシャッフルフェーズを回避します。

これは、バケット化されたテーブルにのみ適用されるバケット結合の概念を、FunctionCatalog に登録された関数によってパーティション化されたテーブルに一般化したものです。ストレージパーティション結合は、現在、互換性のある V2 データソースでサポートされています。

以下の SQL プロパティは、さまざまな最適化を備えたさまざまな結合クエリでストレージパーティション結合を有効にします。

ストレージパーティション結合が実行される場合、クエリプランには結合前の Exchange ノードは含まれません。

以下の例は、ストレージパーティション結合をサポートする Spark V2 データソースである Iceberg （https://iceberg.apache.org/docs/latest/spark-getting-started/）を使用しています。

CREATE TABLE prod.db.target (id INT, salary INT, dep STRING)
USING iceberg
PARTITIONED BY (dep, bucket(8, id))

CREATE TABLE prod.db.source (id INT, salary INT, dep STRING)
USING iceberg
PARTITIONED BY (dep, bucket(8, id))

EXPLAIN SELECT * FROM target t INNER JOIN source s
ON t.dep = s.dep AND t.id = s.id

-- Plan without Storage Partition Join
== Physical Plan ==
* Project (12)
+- * SortMergeJoin Inner (11)
   :- * Sort (5)
   :  +- Exchange (4) // DATA SHUFFLE
   :     +- * Filter (3)
   :        +- * ColumnarToRow (2)
   :           +- BatchScan (1)
   +- * Sort (10)
      +- Exchange (9) // DATA SHUFFLE
         +- * Filter (8)
            +- * ColumnarToRow (7)
               +- BatchScan (6)


SET 'spark.sql.sources.v2.bucketing.enabled' 'true'
SET 'spark.sql.iceberg.planning.preserve-data-grouping' 'true'
SET 'spark.sql.sources.v2.bucketing.pushPartValues.enabled' 'true'
SET 'spark.sql.requireAllClusterKeysForCoPartition' 'false'
SET 'spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled' 'true'

-- Plan with Storage Partition Join
== Physical Plan ==
* Project (10)
+- * SortMergeJoin Inner (9)
   :- * Sort (4)
   :  +- * Filter (3)
   :     +- * ColumnarToRow (2)
   :        +- BatchScan (1)
   +- * Sort (8)
      +- * Filter (7)
         +- * ColumnarToRow (6)
            +- BatchScan (5)