[大数据]Java-Spark系列10-Spark性能调优概述

一.Spark 性能优化概述

首先笔者能力优先，使用Spark有一段时间，如下是笔者的工作经验的总结。

Spark任务运行图:

Spark的优化思路:
一般是从3个层面进行Spark程序的优化:

1. 运行环境优化
2. RDD算子优化
3. 参数微调

二.运行环境优化

2.1 数据本地性

我们知道HDFS的数据文件存储在不同的datanode，一般数据副本数量是3，因为Spark计算的数据量比较大，如果数据不在本节点，需要通过网络去其它的datanode读取数据。

所以此时我们可以通过提高数据本地性，减少网络传输，来达到性能优化的目的。

1. 计算和存储同节点(executor和HDFS的datanode、hbase的region server同节点)
2. executor数目合适: 如果100个数据界定，3个计算节点，就有97份网络传递，所以此种情况可以适当增加计算节点。
3. 适当增加数据副本数量

2.2 数据存储格式

推荐使用列式存储格式: parquet.
parquet存在如下优先:

1. 相同数据类型的数据有很高压缩比
2. Hive主要支持ORC、也支持parquet

三.RDD算子优化

3.1 尽可能复用同一个RDD

每创建一个RDD都会带来性能的开销，尽可能的对同一个RDD做算子操作，而不要频繁创建新的
RDD。

3.2 对多次使用的RDD进行持久化

如果RDD的算子特别多，需要频繁多次操作同一个RDD，最好的办法是将该RDD进行持久化，

四.参数微调

1. num-executors
   参数说明：该参数用于设置每个Executor进程的内存。Executor内存的大小，很多时候直接决定了Spark作业的性能，而且跟常见的JVM OOM异常，也有直接的关联。

2. executor-cores
   参数说明：该参数用于设置每个Executor进程的CPU core数量。

3. driver-memory
   参数说明：该参数用于设置Driver进程的内存。

4. spark.default.parallelism
   参数说明：该参数用于设置每个stage的默认task数量。

5. spark.storage.memoryFraction
   参数说明：该参数用于设置RDD持久化数据在Executor内存中能占的比例，默认是0.6。

6. spark.shuffle.memoryFraction
   参数说明：该参数用于设置shuffle过程中一个task拉取到上个stage的task的输出后，进行聚合操作时能够使用的Executor内存的比例，默认是0.2。

资源参数参考示例：

./bin/spark-submit \
  --master yarn-cluster \
  --num-executors 100 \
  --executor-memory 6G \
  --executor-cores 4 \
  --driver-memory 1G \
  --conf spark.default.parallelism=1000 \
  --conf spark.storage.memoryFraction=0.5 \
  --conf spark.shuffle.memoryFraction=0.3 \

五.数据倾斜

绝大多数task执行得都非常快，但个别task执行极慢。比如，总共有1000个task，997个task都在1分钟之内执行完了，但是剩余两三个task却要一两个小时。这种情况很常见。

数据倾斜图例:

解决数据倾斜一般有如下几种常用方法:

1. 使用Hive ETL预处理数据
   先使用Hive进行预处理数据，也就是使用Hive先计算一层中间数据，Spark从中间层数据开始计算。

2. 过滤少数导致倾斜的key
   如果发生导致倾斜的key非常少，可以将Spark任务拆分为包含 导致倾斜的key的任务和不包含key的任务。

3. sample采样倾斜key单独进行join
   通过采样，提前预估会发生数据倾斜的key，然后将一个join拆分为两个join，其中一个不包含该key，一个只包含该key，最后将结果集进行union。

4. 调整并行度
   调整Shuffle并行度，数据打散

5. 广播小数据集
   适用于一个大表，一个小表
   不用join连接操作，而改用Broadcast变量与map模拟join操作，完全规避shuffle操作
   spark.sql: spark.sql.autoBroadcastJoinThreshold=104857600

6. 增加随机前缀
   对发生倾斜的RDD增加随机前缀
   对另外一个RDD等量扩容
   如果少量的key发生倾斜，可以先过滤出一个单独的RDD，对另外一个RDD同理吹，join之后再合并

六. Spark常用的调优参数

6.1 在内存中缓存数据

Spark SQL可以通过调用Spark.catalog.cachetable (“tableName”)或DataFrame.cache()来使用内存中的columnar格式缓存表。然后Spark SQL将只扫描所需的列，并自动调优压缩以最小化内存使用和GC压力。你可以调用spark.catalog.uncacheTable(“tableName”)从内存中删除表。

内存缓存的配置可以在SparkSession上使用setConf方法或者使用SQL运行SET key=value命令来完成。

6.2 其它配置项

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

6.3 SQL查询连接的hint

join策略提示BROADCAST、MERGE、SHUFFLE_HASH和SHUFFLE_REPLICATE_NL，在将指定的关系加入到另一个关系时，指示Spark对每个指定的关系使用暗示策略。例如，在表’ t1 '上使用BROADCAST提示时，广播加入(广播散列连接或广播嵌套循环联接取决于是否有等值连接键)与t1的构建方面将由火花即使大小的优先表t1的建议的统计配置spark.sql.autoBroadcastJoinThreshold之上。

当连接两端指定了不同的连接策略提示时，Spark会优先考虑BROADCAST提示而不是MERGE提示，优先考虑SHUFFLE_HASH提示而不是SHUFFLE_REPLICATE_NL提示。当双方都指定了BROADCAST提示或SHUFFLE_HASH提示时，Spark将根据连接类型和关系的大小选择构建端。

请注意，不能保证Spark会选择提示中指定的连接策略，因为特定的策略可能不支持所有的连接类型。

-- We accept BROADCAST, BROADCASTJOIN and MAPJOIN for broadcast hint
SELECT /*+ BROADCAST(r) */ * FROM records r JOIN src s ON r.key = s.key

Coalesce hint允许Spark SQL用户控制输出文件的数量，就像Dataset API中的Coalesce、repartition和repartitionByRange一样，它们可以用于性能调优和减少输出文件的数量。COALESCE hint只有一个分区号作为参数。“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

6.4 自适应查询执行

Adaptive Query Execution (AQE)是Spark SQL中的一种优化技术，它利用运行时统计信息来选择最高效的查询执行计划。默认情况下AQE是禁用的。Spark SQL可以使用Spark.SQL.adaptive.enabled的伞配置来控制是否打开/关闭。从Spark 3.0开始，AQE中有三个主要特性，包括合并shuffle后分区、将排序合并连接转换为广播连接以及倾斜连接优化。

6.5 合并分区后重新组合

当spark.sql.adaptive.enabled和spark.sql.adaptive.coalescePartitions.enabled配置都为true时，该特性根据map输出统计信息来合并post shuffle分区。这个特性简化了运行查询时shuffle分区号的调优。您不需要设置合适的shuffle分区号来适合您的数据集。一旦您通过Spark .sql. adaptive.coalescepartitions . initialpartitionnum配置设置了足够大的初始shuffle分区数，Spark就可以在运行时选择适当的shuffle分区号。

6.6 将排序合并联接转换为广播联接

当任何连接侧的运行时统计数据小于广播散列连接阈值时，AQE将排序合并连接转换为广播散列连接。这不是一样有效规划一个广播散列连接首先,但这总比继续做分类合并加入,我们可以节省连接双方的排序,并在本地读取洗牌文件节省网络流量(如果spark.sql.adaptive.localShuffleReader.enabled被设置为true)

6.7 优化倾斜连接

数据倾斜会严重降低连接查询的性能。该特性通过将倾斜任务拆分(如果需要的话还可以复制)为大小大致相同的任务，动态处理排序-合并连接中的倾斜任务。当spark.sql.adaptive.enabled和spark.sql.adaptive.skewJoin.enabled配置同时启用时生效。

参考:

1. http://spark.apache.org/docs/latest/rdd-programming-guide.html
2. https://tech.meituan.com/2016/04/29/spark-tuning-basic.html
3. https://blog.csdn.net/meihao5/article/details/81084876
4. http://spark.apache.org/docs/latest/sql-performance-tuning.html