性能调优

Spark 提供了许多用于调优 DataFrame 或 SQL 工作负载性能的技术。广义上讲，这些技术包括数据缓存、更改数据集分区方式、选择最佳连接策略以及为优化器提供可用于构建更高效执行计划的额外信息。

• 缓存数据
• 调优分区
  • Coalesce 提示
• 利用统计信息
• 优化 Join 策略
  • 自动广播连接
  • Join 策略提示
• 自适应查询执行
  • 合并 Shuffle 后分区
  • 拆分倾斜 Shuffle 分区
  • 将排序合并连接转换为广播连接
  • 将排序合并连接转换为 Shuffle 哈希连接
  • 优化倾斜连接
  • 高级定制
• 存储分区连接

缓存数据

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 的统计信息：

• 数据源：Spark 直接从底层数据源读取的统计信息，例如 Parquet 文件元数据中的计数和最小值/最大值。这些统计信息由底层数据源维护。
• 目录：Spark 从目录（例如 Hive Metastore）读取的统计信息。这些统计信息在您运行 ANALYZE TABLE 时收集或更新。
• 运行时：Spark 在查询运行时自行计算的统计信息。这是自适应查询执行框架的一部分。

缺失或不准确的统计信息将阻碍 Spark 选择最佳计划的能力，并可能导致查询性能不佳。因此，检查提供给 Spark 的统计信息以及它在查询规划和执行期间所做的估计会很有帮助。

• 数据对象统计信息：您可以使用 DESCRIBE EXTENDED 检查表或列上的统计信息。
• 查询计划估计：您可以通过 EXPLAIN COST 或 DataFrame.explain(mode="cost") 检查优化查询计划中的 Spark 成本估计。
• 运行时统计信息：您可以在查询运行时通过 SQL UI 的“Details”部分检查这些统计信息。在计划中查找 Statistics(..., isRuntime=true)。

优化 Join 策略

自动广播连接

Join 策略提示

Join 策略提示，即 BROADCAST、MERGE、SHUFFLE_HASH 和 SHUFFLE_REPLICATE_NL，指示 Spark 在将指定关系与另一个关系连接时，对每个指定关系使用提示的策略。例如，当在表 't1' 上使用 BROADCAST 提示时，即使根据统计信息表 't1' 的大小超过配置 spark.sql.autoBroadcastJoinThreshold，Spark 也会优先使用将 't1' 作为构建侧的广播连接（无论是广播哈希连接还是广播嵌套循环连接，取决于是否存在等值连接键）。

当连接的两侧都指定了不同的连接策略提示时，Spark 优先考虑 BROADCAST 提示，其次是 MERGE 提示，然后是 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

更多详情请参阅Join 提示的文档。

自适应查询执行

自适应查询执行 (AQE) 是 Spark SQL 中的一种优化技术，它利用运行时统计信息选择最有效的查询执行计划，自 Apache Spark 3.2.0 起默认启用。Spark SQL 可以通过 spark.sql.adaptive.enabled 作为总开关来开启和关闭 AQE。

合并 Shuffle 后分区

当 spark.sql.adaptive.enabled 和 spark.sql.adaptive.coalescePartitions.enabled 配置都为 true 时，此功能会根据 Map 输出统计信息合并 shuffle 后分区。此功能简化了运行查询时 shuffle 分区数的调优。您无需设置合适的分区数来适应您的数据集。一旦您通过 spark.sql.adaptive.coalescePartitions.initialPartitionNum 配置设置了足够大的初始 shuffle 分区数，Spark 可以在运行时选择合适的 shuffle 分区数。

拆分倾斜 Shuffle 分区

将排序合并连接转换为广播连接

当任何连接侧的运行时统计信息小于自适应广播哈希连接阈值时，AQE 会将排序合并连接转换为广播哈希连接。这不如一开始就规划广播哈希连接效率高，但比继续排序合并连接要好，因为我们可以避免对连接两侧进行排序，并本地读取 shuffle 文件以节省网络流量（前提是 spark.sql.adaptive.localShuffleReader.enabled 为 true）。

将排序合并连接转换为 Shuffle 哈希连接

当所有 shuffle 后分区都小于 spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold 中配置的阈值时，AQE 会将排序合并连接转换为 shuffle 哈希连接。

优化倾斜连接

数据倾斜会严重降低 Join 查询的性能。此功能通过将倾斜任务拆分（并在需要时复制）成大致均匀大小的任务，来动态处理排序合并连接中的倾斜。它在 spark.sql.adaptive.enabled 和 spark.sql.adaptive.skewJoin.enabled 配置都启用时生效。

高级定制

您可以通过提供自己的成本评估器类或排除 AQE 优化器规则来控制 AQE 的工作细节。

存储分区连接

存储分区连接 (SPJ) 是 Spark SQL 中的一种优化技术，它利用现有的存储布局来避免 Shuffle 阶段。

这是 Bucket Joins 概念的泛化，后者仅适用于桶化表，而 SPJ 适用于通过 FunctionCatalog 中注册的函数进行分区的表。存储分区连接目前支持兼容的 V2 数据源。

以下 SQL 属性通过各种优化，在不同的连接查询中启用存储分区连接。

如果执行存储分区连接，查询计划在连接之前将不包含 Exchange 节点。

以下示例使用 Iceberg (https://iceberg.apache.org/docs/latest/spark-getting-started/)，这是一个支持存储分区连接的 Spark V2 数据源。

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)