性能调优

• 将数据缓存到内存中
• 其他配置选项
• SQL 查询的连接策略提示
• SQL 查询的合并提示
• 自适应查询执行
  • 合并后洗牌分区
  • 拆分倾斜洗牌分区
  • 将排序合并连接转换为广播连接
  • 将排序合并连接转换为洗牌哈希连接
  • 优化倾斜连接
  • 杂项

对于某些工作负载，可以通过将数据缓存到内存中或启用一些实验性选项来提高性能。

将数据缓存到内存中

Spark SQL 可以使用内存中的列式格式缓存表，方法是调用 spark.catalog.cacheTable("tableName") 或 dataFrame.cache()。然后 Spark SQL 将仅扫描所需的列，并会自动调整压缩以最大程度地减少内存使用量和 GC 压力。您可以调用 spark.catalog.uncacheTable("tableName") 或 dataFrame.unpersist() 从内存中删除表。

内存缓存的配置可以通过 SparkSession 上的 setConf 方法完成，也可以通过使用 SQL 运行 SET key=value 命令完成。

其他配置选项

以下选项也可用于调整查询执行的性能。随着更多优化自动执行，这些选项可能会在将来的版本中被弃用。

SQL 查询的连接策略提示

连接策略提示，即 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

有关更多详细信息，请参阅 连接提示 的文档。

SQL 查询的合并提示

合并提示允许 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;

有关更多详细信息，请参阅 分区提示 的文档。

自适应查询执行

自适应查询执行 (AQE) 是 Spark SQL 中的一种优化技术，它利用运行时统计信息来选择最有效的查询执行计划，该技术从 Apache Spark 3.2.0 开始默认启用。Spark SQL 可以通过 spark.sql.adaptive.enabled 作为伞形配置来启用和禁用 AQE。从 Spark 3.0 开始，AQE 中有三个主要功能：包括合并后洗牌分区、将排序合并连接转换为广播连接以及倾斜连接优化。

合并后洗牌分区

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

拆分倾斜洗牌分区

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

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

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

当所有后混洗分区都小于阈值时，AQE 会将排序合并连接转换为混洗哈希连接，最大阈值可以查看配置 spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold。

优化倾斜连接

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

杂项