转载请注明出处，谢谢合作～

该篇中的示例暂时只有 Scala 版本～

性能调优

• 在内存中缓存数据（Caching Data In Memory）
• 其他配置项（Other Configuration Options）
• SQL 查询连接策略提示（Join Strategy Hints for SQL Queries）
• SQL 查询合并提示（Coalesce Hints for SQL Queries）
• 自适应查询引擎（Adaptive Query Execution）
  • 合并 shuffle 分区（Coalescing Post Shuffle Partitions）
  • 将 sort-merge 连接转换为 broadcast 连接（Converting sort-merge join to broadcast join）
  • 优化连接中的数据倾斜（Optimizing Skew Join）

对于某些工作场景，可以通过在内存中缓存数据或者开启一些试验功能来提升性能。

在内存中缓存数据

Spark SQL 能做以一种内存中的列式存储的格式缓存数据，可以通过调用 spark.catalog.cacheTable("tableName") 或者 dataFrame.cache() 方法实现。之后 Spark SQL 只需要扫描用到的列中的数据，而且能够自动使用压缩来节省内存空间，缓解 GC 压力。可以通过调用 spark.catalog.uncacheTable("tableName") 来删除缓存。

内存缓存的配置项可以通过 SparkSession 对象的 setConf 方法或者执行 SQL 中的 SET key=value 语句来设置。

其他配置项

下列选项也可以用来对查询的性能进行调优。由于越来越多的优化都会自动进行，这些选项在未来的版本中可能被被废弃。

SQL 查询连接策略提示

目前的连接策略提示有 BROADCAST，MERGE，SHUFFLE_HASH 和 SHUFFLE_REPLICATE_NL，在一张关系表和另一张关系表做连接操作时，Spark 会根据连接策略提示采用不同的连接策略。例如，当对表「t1」使用 BROADCAST 连接策略提示时，Spark 会优先采用广播连接（根据是否含有 equi-join 键来决定采用 broadcast hash join 或者 broadcast nested loop join），即使的统计信息表明其大小超出了上述配置 spark.sql.autoBroadcastJoinThreshold 的限值。

如果连接操作的调用者和被调用者采用了不同的连接策略提示，优先级从高到低为：BROADCAST、MERGE、SHUFFLE_HASH、SHUFFLE_REPLICATE_NL。如果连接操作的调用者和被调用者都采用了 BROADCAST 策略提示或者 SHUFFLE_HASH 策略提示，Spark 会根据连接类型和表数据量大小做出选择。

注意，由于指定的连接策略提示可能并不支持所有的连接类型，所以并不能保证 Spark 所采取的连接策略一定是被指定的那个。

spark.table("src").join(spark.table("records").hint("broadcast"), "key").show()

详情参见文档 Join Hints。

SQL 查询合并提示

合并提示可以让 Spark SQL 用户控制一些算子的输出文件数量，这些算子包括 coalesce，repartition 和 repartitionByRange。合并指引可以用来调优性能，减少输出文件的数量。「COALESCE」指引只有一个分区数量作为参数。「REPARTITION」的参数可以是分区数或者列名或者两者一起。「REPARTITION_BY_RANGE」提示必须有一个列名参数，分区数是可选的参数。

SELECT /*+ COALESCE(3) */ * FROM t
SELECT /*+ REPARTITION(3) */ * FROM t
SELECT /*+ REPARTITION(c) */ * FROM t
SELECT /*+ REPARTITION(3, c) */ * FROM t
SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t
SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t

详情参见文档 Partitioning Hints。

自适应查询引擎

自适应查询引擎（AQE）是 Spark SQL 当中的一项优化技术，能够充分利用运行时统计信息来选取最高效的查询计划。AQE 默认是关闭的，Spark SQL 可以使用保护伞配置参数 spark.sql.adaptive.enabled 来控制是否开启。Spark 3.0 中的 AQE 主要有三种优化手段，包括合并 shuffle 分区，将 sort-merge 连接转换为 broadcast 连接，优化连接中的数据倾斜。

合并 shuffle 分区

当参数 spark.sql.adaptive.enabled 和 spark.sql.adaptive.coalescePartitions.enabled 都被开启时，通过 map 端的统计信息合并 reduce 端的分区数。这种方式简化了查询中 shuffle 分区数量的调优，不需要针对数据指定一个合理的分区数，一旦通过参数 spark.sql.adaptive.coalescePartitions.initialPartitionNum 设置了一个足够大的初始 shuffle 分区数，Spark 会在运行时选取合理的分区数。

将 sort-merge 连接转换为 broadcast 连接

当连接操作的调用者或者被调用者的数据量小于广播哈希连接的限值时，AQE 会将 sort-merge 连接转换为 broadcast 连接。这种方式并不比在开始就指定广播连接策略更高效，但是也比 sort-merge 连接好多了。优化之后可以在一个节点上保存需要连接的所有数据，读取本地文件节省了网络传输开销（需要参数 spark.sql.adaptive.localShuffleReader.enabled 设置为 true）。

优化连接中的数据倾斜

数据倾斜会严重降低连接查询的效率，AQE 通过分割（或者折叠）将倾斜的数据分区组合成大小基本均匀的子任务来动态处理 sort-merge 连接中的数据倾斜。当参数spark.sql.adaptive.enabled 和 spark.sql.adaptive.skewJoin.enabled 都开启时生效。