性能调优

• 在内存中缓存数据
• 其他配置选项
• SQL 查询的 Join 策略提示
• SQL 查询的 Coalesce 提示
• 自适应查询执行
  • 合并 Shuffle 后的分区
  • 分割倾斜的 shuffle 分区
  • 将 sort-merge join 转换为 broadcast join
  • 将 sort-merge join 转换为 shuffled hash 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 策略提示

连接策略提示，即 BROADCAST、MERGE、SHUFFLE_HASH 和 SHUFFLE_REPLICATE_NL，指示 Spark 在将每个指定的关联与另一个关联进行连接时，在每个指定的关联上使用提示的策略。 例如，当在表“t1”上使用 BROADCAST 提示时，以“t1”作为构建侧的广播连接（广播哈希连接或广播嵌套循环连接，具体取决于是否存在任何等值连接键）将被 Spark 优先考虑，即使统计信息建议的表“t1”的大小高于配置 spark.sql.autoBroadcastJoinThreshold。

如果在连接的两侧指定了不同的连接策略提示，则 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 Hints 的文档。

SQL 查询的 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 的文档。

自适应查询执行

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

合并 Shuffle 后的分区

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

分割倾斜的 shuffle 分区

将 sort-merge join 转换为 broadcast join

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

将 sort-merge join 转换为 shuffled hash join

当所有 shuffle 后的分区都小于阈值时，AQE 会将排序合并连接转换为 shuffled hash join，最大阈值可以参考配置 spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold。

优化倾斜 Join

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

其他