性能调优

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

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

缓存数据

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;

更多详细信息，请参考 分区提示 (Partitioning Hints) 文档。

利用统计信息

Apache Spark 在众多可能的选项中选择最佳执行计划的能力，部分取决于它对执行计划中每个节点（读取、过滤、连接等）将输出多少行的估计。这些估计又基于通过多种方式提供给 Spark 的统计信息：

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

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

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

优化连接策略

自动广播连接

连接策略提示

连接策略提示，即 BROADCAST、MERGE、SHUFFLE_HASH 和 SHUFFLE_REPLICATE_NL，指示 Spark 在将指定的某个关系与另一个关系连接时使用该提示策略。例如，当在表“t1”上使用 BROADCAST 提示时，即使统计信息建议的表“t1”的大小超过了配置 spark.sql.autoBroadcastJoinThreshold，Spark 也会优先使用以“t1”为构建侧（build side）的广播连接（取决于是否存在等值连接键，可能是广播哈希连接或广播嵌套循环连接）。

当在连接的两侧指定了不同的连接策略提示时，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 src s JOIN records r ON s.key = r.key

更多详细信息，请参考 连接提示 (Join Hints) 文档。

自适应查询执行 (AQE)

自适应查询执行 (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 分区

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

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

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

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

优化倾斜连接

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

高级自定义

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

存储分区连接 (Storage Partition Join)

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

这是桶连接 (Bucket Joins) 概念的推广。桶连接仅适用于 分桶 (bucketed) 表，而存储分区连接适用于由 FunctionCatalog 中注册的函数进行分区的表。存储分区连接目前支持兼容的 V2 数据源。

以下 SQL 属性支持在具有各种优化的不同连接查询中启用存储分区连接。

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

以下示例使用了 Iceberg (https://iceberg.org.cn/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)