从头捋了一遍Spark性能优化经验，我不信你全会

Spark是大数据分析的利器，在工作中用到spark的地方也比较多，这篇总结是希望能将自己使用spark的一些调优经验分享出来。

1. 避免创建重复的RDD，尽量复用同一份数据。

2. 尽量避免使用shuffle类算子，因为shuffle操作是spark中消耗性能的地方，reduceByKey、join、distinct、repartition等算子都会触发shuffle操作，尽量使用map类的非shuffle算子

3. 用aggregateByKey和reduceByKey替代groupByKey,因为前两个是预聚合操作，会在每个节点本地对相同的key做聚合，等其他节点拉取所有节点上相同的key时，会大大减少磁盘IO以及网络开销。

4. repartition适用于RDD[V], partitionBy适用于RDD[K, V]

5. mapPartitions操作替代普通map，foreachPartitions替代foreach

6. filter操作之后进行coalesce操作，可以减少RDD的partition数量

7. 如果有RDD复用，尤其是该RDD需要花费比较长的时间，建议对该RDD做cache，若该RDD每个partition需要消耗很多内存，建议开启Kryo序列化机制(据说可节省2到5倍空间),若还是有比较大的内存开销，可将storage_level设置为MEMORY_AND_DISK_SER

8. 尽量避免在一个Transformation中处理所有的逻辑，尽量分解成map、filter之类的操作

9. 多个RDD进行union操作时，避免使用rdd.union(rdd).union(rdd).union(rdd)这种多重union，rdd.union只适合2个RDD合并，合并多个时采用SparkContext.union(Array(RDD))，避免union嵌套层数太多，导致的调用链路太长，耗时太久，且容易引发StackOverFlow

10. spark中的Group/join/XXXByKey等操作，都可以指定partition的个数，不需要额外使用repartition和partitionBy函数

11. 尽量保证每轮Stage里每个task处理的数据量>128M

12. 如果2个RDD做join，其中一个数据量很小，可以采用Broadcast Join，将小的RDD数据collect到driver内存中，将其BroadCast到另外以RDD中，其他场景想优化后面会讲

13. 2个RDD做笛卡尔积时，把小的RDD作为参数传入，如BigRDD.certesian(smallRDD)

14. 若需要Broadcast一个大的对象到远端作为字典查询，可使用多executor-cores，大executor-memory。若将该占用内存较大的对象存储到外部系统，executor-cores=1， executor-memory=m(默认值2g),可以正常运行，那么当大字典占用空间为size(g)时，executor-memory为2*size，executor-cores=size/m(向上取整)

15.如果对象太大无法BroadCast到远端，且需求是根据大的RDD中的key去索引小RDD中的key，可使用zipPartitions以hash join的方式实现，具体原理参考下一节的shuffle过程

16. 如果需要在repartition重分区之后还要进行排序，可直接使用repartitionAndSortWithinPartitions，比分解操作效率高，因为它可以一边shuffle一边排序

spark中的shuffle操作功能：将分布在集群中多个节点上的同一个key，拉取到同一个节点上，进行聚合或join操作，类似洗牌的操作。这些分布在各个存储节点上的数据重新打乱然后汇聚到不同节点的过程就是shuffle过程。

RDD的特性是不可变的带分区的记录集合，Spark提供了Transformation和Action两种操作RDD的方式。Transformation是生成新的RDD，包括map, flatMap, filter, union, sample, join, groupByKey, cogroup, ReduceByKey, cros, sortByKey, mapValues等；Action只是返回一个结果，包括collect，reduce，count，save，lookupKey等

Spark所有的算子操作中是否使用shuffle过程要看计算后对应多少分区：

shuffle过程分为shuffle write和shuffle read两部分

shuffle过程中，各个节点会通过shuffle write过程将相同key都会先写入本地磁盘文件中，然后其他节点的shuffle read过程通过网络传输拉取各个节点上的磁盘文件中的相同key。这其中大量数据交换涉及到的网络传输和文件读写操作是shuffle操作十分耗时的根本原因

参数spark.shuffle.manager用于设置ShuffleManager的类型。Spark1.5以后，该参数有三个可选项：hash、sort和tungsten-sort。HashShuffleManager是Spark1.2以前的默认值，Spark1.2之后的默认值都是SortShuffleManager。tungsten-sort与sort类似，但是使用了tungsten计划中的堆外内存管理机制，内存使用效率更高。

由于SortShuffleManager默认会对数据进行排序，因此如果业务需求中需要排序的话，使用默认的SortShuffleManager就可以；但如果不需要排序，可以通过bypass机制或设置HashShuffleManager避免排序，同时也能提供较好的磁盘读写性能。

HashShuffleManager流程：

SortShuffleManager流程：

bypass机制通过参数spark.shuffle.sort.bypassMergeThreshold设置，默认值是200，表示当ShuffleManager是SortShuffleManager时，若shuffle read task的数量小于这个阈值（默认200）时，则shuffle write过程中不会进行排序操作，而是直接按照未经优化的HashShuffleManager的方式写数据，但后会将每个task产生的所有临时磁盘文件合并成一个文件，并创建索引文件。

这里给出的调优建议是，当使用SortShuffleManager时，如果的确不需要排序，可以将这个参数值调大一些，大于shuffle read task的数量。那么此时就会自动开启bypass机制，map-side就不会进行排序了，减少排序的性能开销，提升shuffle操作效率。但这种方式并没有减少shuffle write过程产生的磁盘文件数量，所以写的性能没有改变。

如果使用HashShuffleManager，可以设置spark.shuffle.consolidateFiles参数。该参数默认为false，只有当使用HashShuffleManager且该参数设置为True时，才会开启consolidate机制，大幅度合并shuffle write过程产生的输出文件，对于shuffle read task 数量特别多的情况下，可以极大地减少磁盘IO开销，提升shuffle性能。参考社区同学给出的数据，consolidate性能比开启bypass机制的SortShuffleManager高出10% ~ 30%。

除了上述的几个参数调优，shuffle过程还有一些参数可以提高性能：

SparkSQL中的shuffle错误：

RDD中的shuffle错误：

处理shuffle类操作的注意事项：

Spark所有的操作中，join操作是复杂、代价大的操作，也是大部分业务场景的性能瓶颈所在。所以针对join操作的优化是使用spark必须要学会的技能。

spark的join操作也分为Spark SQL的join和Spark RDD的join。

Hash Join

Hash Join的执行方式是先将小表映射成Hash Table的方式，再将大表使用相同方式映射到Hash Table，在同一个hash分区内做join匹配。

hash join又分为broadcast hash join和shuffle hash join两种。其中Broadcast hash join，顾名思义，就是把小表广播到每一个节点上的内存中，大表按Key保存到各个分区中，小表和每个分区的大表做join匹配。这种情况适合一个小表和一个大表做join且小表能够在内存中保存的情况。如下图所示：

当Hash Join不能适用的场景就需要Shuffle Hash Join了，Shuffle Hash Join的原理是按照join Key分区，key相同的数据必然分配到同一分区中，将大表join分而治之，变成小表的join，可以提高并行度。执行过程也分为两个阶段：

• shuffle阶段：分别将两个表按照join key进行分区，将相同的join key数据重分区到同一节点；

• hash join阶段：每个分区节点上的数据单独执行单机hash join算法。

Shuffle Hash Join的过程如下图所示：

SparkSQL针对两张大表join的情况提供了全新的算法——Sort-merge join，整个过程分为三个步骤：

Spark的RDD join没有上面这么多的分类，但是面临的业务需求是一样的。如果是大表join小表的情况，则可以将小表声明为broadcast变量，使用map操作快速实现join功能，但又不必执行Spark core中的join操作。

如果是两个大表join，则必须依赖Spark Core中的join操作了。Spark RDD Join的过程可以自行阅读源码了解，这里只做一个大概的讲解。

spark的join过程中核心的函数是cogroup方法，这个方法中会判断join的两个RDD所使用的partitioner是否一样，如果分区相同，即存在OneToOneDependency依赖，不用进行hash分区，可直接join；如果要关联的RDD和当前RDD的分区不一致时，就要对RDD进行重新hash分区，分到正确的分区中，即存在ShuffleDependency，需要先进行shuffle操作再join。因此提升join效率的一个思路就是使得两个RDD具有相同的partitioners。

所以针对Spark RDD的join操作的优化建议是：

均匀数据分布的情况下，前面所说的优化建议就足够了。但存在数据倾斜时，仍然会有性能问题。主要体现在绝大多数task执行得都非常快，个别task执行很慢，拖慢整个任务的执行进程，甚至可能因为某个task处理的数据量过大而爆出OOM错误。

shuffle操作中需要将各个节点上相同的key拉取到某一个节点上的一个task处理，如果某个key对应的数据量特别大，就会发生数据倾斜。

分析数据分布

如果是Spark SQL中的group by、join语句导致的数据倾斜，可以使用SQL分析执行SQL中的表的key分布情况；如果是Spark RDD执行shuffle算子导致的数据倾斜，可以在Spark作业中加入分析Key分布的代码，使用countByKey()统计各个key对应的记录数。

数据倾斜的解决方案

这里参考美团技术博客中给出的几个方案。

1、报错信息

解决方案：

上面这段错误提示的本质是Linux操作系统无法创建更多进程，导致出错，并不是系统的内存不足。因此要解决这个问题需要修改Linux允许创建更多的进程，就需要修改Linux大进程数

2、报错信息

由于Spark在计算的时候会将中间结果存储到/tmp目录，而目前linux又都支持tmpfs，其实就是将/tmp目录挂载到内存当中, 那么这里就存在一个问题，中间结果过多导致/tmp目录写满而出现如下错误

No Space Left on the device（Shuffle临时文件过多）

解决方案：

修改配置文件spark-env.sh,把临时文件引入到一个自定义的目录中去, 即:

export SPARK_LOCAL_DIRS=/home/utoken/datadir/spark/tmp

3、报错信息

Worker节点中的work目录占用许多磁盘空间, 这些是Driver上传到worker的文件, 会占用许多磁盘空间

解决方案：

需要定时做手工清理work目录

4、spark-shell提交Spark Application如何解决依赖库

解决方案：

利用–driver-class-path选项来指定所依赖的jar文件，注意的是–driver-class-path后如果需要跟着多个jar文件的话，jar文件之间使用冒号:来分割。

5、内存不足或数据倾斜导致Executor Lost，shuffle fetch失败，Task重试失败等（spark-submit提交）

解决方案：

增加worker内存，或者相同资源下增加partition数目，这样每个task要处理的数据变少，占用内存变少

如果存在shuffle过程，设置shuffle read阶段的并行数