Spark 优化 (一) --------- Spark 性能调优

布满荆棘的人生 2024-04-06 10:47 87阅读 0赞

#### 目录 ####

*  一、常规性能调优
 *   *  1. 最优资源配置
     *  2. RDD优化
     *  3. 并行度调节
     *  4. 广播大变量
     *  5. Kryo序列化
     *  6. 调节本地化等待时长
 *  二、算子调优
 *   *  1. mapPartitions
     *  2. foreachPartition 优化数据库操作
     *  3. filter 与 coalesce 的配合使用
     *  4. repartition解决SparkSQL低并行度问题
     *  5. reduceByKey 预聚合
 *  三、Shuffle调优
 *   *  1. 调节map端缓冲区大小
     *  2. 调节reduce端拉取数据缓冲区大小
     *  3. 调节reduce端拉取数据重试次数
     *  4. 调节reduce端拉取数据等待间隔
     *  5. 调节SortShuffle排序操作阈值
 *  四、JVM调优
 *   *  1. 降低cache操作的内存占比
     *  2. 调节Executor堆外内存
     *  3. 调节连接等待时长

--------------------

## 一、常规性能调优 ##

### 1. 最优资源配置 ###

Spark性能调优的第一步，就是为任务分配更多的资源，在一定范围内，增加资源的分配与性能的提升是成正比的，实现了最优的资源配置后，在此基础上再考虑进行后面论述的性能调优策略。资源的分配在使用脚本提交Spark任务时进行指定，标准的Spark任务提交脚本如下所示：

bin/spark-submit \
    --class com.fancyry.spark.Analysis \
    --master yarn--deploy-mode cluster
    --num-executors 80 \
    --driver-memory 6g \
    --executor-memory 6g \
    --executor-cores 3 \
    /usr/opt/modules/spark/jar/spark.jar \

可以进行分配的资源如表所示：

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>–num-executors</td> 
   <td>配置Executor的数量</td> 
  </tr> 
  <tr> 
   <td>–driver-memory</td> 
   <td>配置Driver内存（影响不大）</td> 
  </tr> 
  <tr> 
   <td>–executor-memory</td> 
   <td>配置每个Executor的内存大小</td> 
  </tr> 
  <tr> 
   <td>–executor-cores</td> 
   <td>配置每个Executor的CPU core数量</td> 
  </tr> 
 </tbody> 
</table>

调节原则：尽量将任务分配的资源调节到可以使用的资源的最大限度。对于具体资源的分配，我们分别讨论Spark的两种Cluster运行模式：

➢第一种是SparkStandalone模式，你在提交任务前，一定知道或者可以从运维部门获取到你可以使用的资源情况，在编写submit脚本的时候，就根据可用的资源情况进行资源的分配，比如说集群有15台机器，每台机器为8G内存，2个CPU core，那么就指定15个Executor，每个Executor分配8G内存，2个CPUcore。

➢第二种是SparkYarn模式，由于Yarn使用资源队列进行资源的分配和调度，在编写submit脚本的时候，就根据Spark作业要提交到的资源队列，进行资源的分配，比如资源队列有400G内存，100个CPU core，那么指定50个Executor，每个Executor分配8G内存，2个CPU core。

对各项资源进行了调节后，得到的性能提升会有如下表现：

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>解析</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>增加Executor·个数</td> 
   <td>在资源允许的情况下，增加Executor的个数可以提高执行task的并行度。比如有4个Executor，每个Executor有2个CPU core，那么可以并行执行8个task，如果将Executor的个数增加到8个（资源允许的情况下），那么可以并行执行16个task，此时的并行能力提升了一倍。</td> 
  </tr> 
  <tr> 
   <td>增加每个Executor的CPU core个数</td> 
   <td>在资源允许的情况下，增加每个Executor的Cpu core个数，可以提高执行task的并行度。比如有4个Executor，每个Executor有2个CPU core，那么可以并行执行8个task，如果将每个Executor的CPU core个数增加到4个 (资源允许的情况下)，那么可以并行执行16个task，此时的并行能力提升了一倍。</td> 
  </tr> 
  <tr> 
   <td>增加每个Executor的内存量</td> 
   <td>在资源允许的情况下，增加每个Executor的内存量以后，对性能的提升有三点：1.可以缓存更多的数据(即对RDD进行cache)，写入磁盘的数据相应减少，甚至可以不写入磁盘，减少了可能的磁盘IO；2.可以为shuffle操作提供更多内存，即有更多空间来存放reduce端拉取的数据，写入磁盘的数据相应减少，甚至可以不写入磁盘，减少了可能的磁盘IO；3.可以为task的执行提供更多内存，在task的执行过程中可能创建很多对象，内存较小时会引发频繁的GC，增加内存后，可以避免频繁的GC，提升整体性能。</td> 
  </tr> 
 </tbody> 
</table>

补充：生产环境 Spark submit 脚本配置

bin/spark-submit \
    --class com.fancy.spark.WordCount \
    --master yarn \
    --deploy-modecluster \
    --num-executors 80 \
    --driver-memory 6g \
    --executor-memory 6g \
    --executor-cores 3 \
    --queue root.default \
    --conf spark.yarn.executor.memoryOverhead=2048 \
    --conf spark.core.connection.ack.wait.timeout=300 \/usr/local/spark/spark.jar

**参数配置参考值**：

![在这里插入图片描述][a7c9bd51118848e882a7bfedc123b377.png]

### 2. RDD优化 ###

**A、RDD复用**

在对 RDD 进行算子时，要避免相同的算子和计算逻辑之下对 RDD 进行重复的计算

![在这里插入图片描述][7e8740fc29424e5d8859b7249ccbd3f9.png]  
对上图中的RDD计算架构进行修改，得到如下图所示的优化结果：

![在这里插入图片描述][5276900395354843ad6d341a3b8cc3c0.png]

**B、RDD持久化**

在Spark中，当多次对同一个RDD执行算子操作时，每一次都会对这个RDD以之前的父RDD重新计算一次，这种情况是必须要避免的，对同一个RDD的重复计算是对资源的极大浪费，因此，必须对多次使用的RDD进行持久化，通过持久化将公共RDD的数据缓存到内存/磁盘中，之后对于公共RDD的计算都会从内存/磁盘中直接获取RDD数据。对于RDD的持久化，有两点需要说明：

➢RDD的持久化是可以进行序列化的，当内存无法将RDD的数据完整的进行存放的时候，可以考虑使用序列化的方式减小数据体积，将数据完整存储在内存中。

➢如果对于数据的可靠性要求很高，并且内存充足，可以使用副本机制，对RDD数据进行持久化。当持久化启用了复本机制时，对于持久化的每个数据单元都存储一个副本，放在其他节点上面，由此实现数据的容错，一旦一个副本数据丢失，不需要重新计算，还可以使用另外一个副本。

**C、RDD尽可能早的filter操作**

获取到初始RDD后，应该考虑尽早地过滤掉不需要的数据，进而减少对内存的占用，从而提升Spark作业的运行效率。

### 3. 并行度调节 ###

Spark作业中的并行度指各个stage的task的数量。如果并行度设置不合理而导致并行度过低，会导致资源的极大浪费，例如，20个Executor，每个Executor分配3个CPU core，而Spark作业有40个task，这样每个Executor分配到的task个数是2个，这就使得每个Executor有一个CPU core空闲，导致资源的浪费。

理想的并行度设置，应该是让并行度与资源相匹配，简单来说就是在资源允许的前提下，并行度要设置的尽可能大，达到可以充分利用集群资源。合理的设置并行度，可以提升整个Spark作业的性能和运行速度。

Spark官方推荐，task数量应该设置为Spark作业总CPU core数量的2~3倍。之所以没有推荐task数量与CPU core总数相等，是因为task的执行时间不同，有的task执行速度快而有的task执行速度慢，如果task数量与CPU core总数相等，那么执行快的task执行完成后，会出现CPU core空闲的情况。如果task数量设置为CPU core总数的2~3倍，那么一个task执行完毕后，CPU core会立刻执行下一个task，降低了资源的浪费，同时提升了Spark作业运行的效率。

Spark作业并行度的设置如下所示：

valconf = new SparkConf().set("spark.default.parallelism", "500")

### 4. 广播大变量 ###

默认情况下，task中的算子中如果使用了外部的变量，每个task都会获取一份变量的复本，这就造成了内存的极大消耗。

一方面，如果后续对RDD进行持久化，可能就无法将RDD数据存入内存，只能写入磁盘，磁盘IO将会严重消耗性能；另一方面，task在创建对象的时候，也许会发现堆内存无法存放新创建的对象，这就会导致频繁的GC，GC会导致工作线程停止，进而导致Spark暂停工作一段时间，严重影响Spark性能。

假设当前任务配置了20个Executor，指定500个task，有一个20M的变量被所有task共用，此时会在500个task中产生500个副本，耗费集群10G的内存，如果使用了广播变量，那么每个Executor保存一个副本，一共消耗400M内存，内存消耗减少了5倍。广播变量在每个Executor保存一个副本，此Executor的所有task共用此广播变量，这让变量产生的副本数量大大减少。

在初始阶段，广播变量只在Driver中有一份副本。task在运行的时候，想要使用广播变量中的数据，此时首先会在自己本地的Executor对应的BlockManager中尝试获取变量，如果本地没有，BlockManager就会从Driver或者其他节点的BlockManager上远程拉取变量的复本，并由本地的BlockManager进行管理；之后此Executor的所有task都会直接从本地的BlockManager中获取变量。

### 5. Kryo序列化 ###

默认情况下，Spark使用Java的序列化机制。Java的序列化机制使用方便，不需要额外的配置，在算子中使用的变量实现Serializable接口即可，但是，Java序列化机制的效率不高，序列化速度慢并且序列化后的数据所占用的空间依然较大。

Kryo 序列化机制比 Java 序列化机制性能提高 10 倍左右，Spark 之所以没有默认使用 Kryo 作为序列化类库，是因为它不支持所有对象的序列化，同时 Kryo 需要用户在使用前注册需要序列化的类型，不够方便，但从 Spark 2.0.0 版本开始，简单类型、简单类型数组、字符串类型的Shuffling RDDs 已经默认使用Kryo序列化方式了。

public class MyKryoRegistrator implements KryoRegistrator {
         
         @Overridepublic void registerClasses(Kryo kryo) {
        
         	kryo.register(StartupReportLogs.class);
         }
    }

配置Kryo序列化方式的实例代码：
    
    //创建SparkConf对象
    val conf = new SparkConf().setMaster(...).setAppName(...)
    
    //使用Kryo序列化库，如果要使用Java序列化库，需要把该行屏蔽掉
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");  
    
    //在Kryo序列化库中注册自定义的类集合，如果要使用Java序列化库，需要把该行屏蔽掉
    conf.set("spark.kryo.registrator", "fancy.com.MyKryoRegistrator");

### 6. 调节本地化等待时长 ###

Spark 作业运行过程中，Driver 会对每一个 stage 的 task 进行分配。根据 Spark 的 task 分配算法，Spark 希望 task 能够运行在它要计算的数据算在的节点 (数据本地化思想)，这样就可以避免数据的网络传输。通常来说，task可能不会被分配到它处理的数据所在的节点，因为这些节点可用的资源可能已经用尽，此时，Spark会等待一段时间，默认3s，如果等待指定时间后仍然无法在指定节点运行，那么会自动降级，尝试将task分配到比较差的本地化级别所对应的节点上，比如将task分配到离它要计算的数据比较近的一个节点，然后进行计算，如果当前级别仍然不行，那么继续降级。

当 task 要处理的数据不在 task 所在节点上时，会发生数据的传输。task 会通过所在节点的 BlockManager 获取数据，BlockManager 发现数据不在本地时，户通过网络传输组件从数据所在节点的BlockManager 处获取数据。网络传输数据的情况是我们不愿意看到的，大量的网络传输会严重影响性能，因此，我们希望通过调节本地化等待时长，如果在等待时长这段时间内，目标节点处理完成了一部分 task，那么当前的 task 将有机会得到执行，这样就能够改善 Spark 作业的整体性能。Spark 的本地化等级如表所示

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>解析</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>PROCESS_LOCAL</td> 
   <td>进程本地化，task和数据在同一个Executor中，性能最好。</td> 
  </tr> 
  <tr> 
   <td>NODE_LOCAL</td> 
   <td>节点本地化，task和数据在同一个节点中，但是task和数据不在同一个Executor中，数据需要在进程间进行传输。</td> 
  </tr> 
  <tr> 
   <td>RACK_LOCAL</td> 
   <td>机架本地化，task和数据在同一个机架的两个节点上，数据需要通过网络在节点之间进行传输。</td> 
  </tr> 
  <tr> 
   <td>NO_PREF</td> 
   <td>对于task来说，从哪里获取都一样，没有好坏之分。</td> 
  </tr> 
  <tr> 
   <td>ANY</td> 
   <td>task和数据可以在集群的任何地方，而且不在一个机架中，性能最差。</td> 
  </tr> 
 </tbody> 
</table>

在Spark项目开发阶段，可以使用client模式对程序进行测试，此时，可以在本地看到比较全的日志信息，日志信息中有明确的task数据本地化的级别，如果大部分都是 PROCESS\_LOCAL，那么就无需进行调节，但是如果发现很多的级别都是 NODE\_LOCAL、ANY，那么需要对本地化的等待时长进行调节，通过延长本地化等待时长，看看 task 的本地化级别有没有提升，并观察 Spark 作业的运行时间有没有缩短。注意，过犹不及，不要将本地化等待时长延长地过长，导致因为大量的等待时长，使得 Spark作业的运行时间反而增加了。Spark本地化等待时长的设置如代码所示：

val conf = new SparkConf().set("spark.locality.wait", "6")

## 二、算子调优 ##

### 1. mapPartitions ###

普通的map算子对RDD中的每一个元素进行操作，而mapPartitions算子对RDD中每一个分区进行操作。如果是普通的map算子，假设一个partition有1万条数据，那么map算子中的function要执行1万次，也就是对每个元素进行操作。

![在这里插入图片描述][9b60409a3345445492645336fcbb5bdc.png]  
如果是 mapPartition 算子，由于一个 task 处理一个 RDD 的 partition，那么一个task只会执行一次function，function 一次接收所有的 partition 数据，效率比较高。

![在这里插入图片描述][5fa3d8d700e44a28ab16ca6d02004b0e.png]

比如，当要把RDD中的所有数据通过JDBC写入数据，如果使用map算子，那么需要对RDD中的每一个元素都创建一个数据库连接，这样对资源的消耗很大，如果使用mapPartitions算子，那么针对一个分区的数据，只需要建立一个数据库连接。

mapPartitions算子也存在一些缺点：对于普通的map操作，一次处理一条数据，如果在处理了2000条数据后内存不足，那么可以将已经处理完的2000条数据从内存中垃圾回收掉；但是如果使用mapPartitions算子，但数据量非常大时，function一次处理一个分区的数据，如果一旦内存不足，此时无法回收内存，就可能会OOM，即内存溢出。

因此，mapPartitions算子适用于数据量不是特别大的时候，此时使用mapPartitions算子对性能的提升效果还是不错的。（当数据量很大的时候，一旦使用mapPartitions算子，就会直接OOM）在项目中，应该首先估算一下RDD的数据量、每个partition的数据量，以及分配给每个Executor的内存资源，如果资源允许，可以考虑使用mapPartitions算子代替map。

### 2. foreachPartition 优化数据库操作 ###

在生产环境中，通常使用 foreachPartition 算子来完成数据库的写入，通过 foreachPartition 算子的特性，可以优化写数据库的性能。

如果使用 foreach 算子完成数据库的操作，由于 foreach 算子是遍历RDD的每条数据，因此，每条数据都会建立一个数据库连接，这是对资源的极大浪费，因此，对于写数据库操作，我们应当使用 foreachPartition 算子。

与 mapPartitions 算子非常相似，foreachPartition 是将RDD 的每个分区作为遍历对象，一次处理一个分区的数据，也就是说，如果涉及数据库的相关操作，一个分区的数据只需要创建一次数据库连接，如图所示：

![在这里插入图片描述][fd61c26f5fa74c2d8f7cde5829162f63.png]  
使用了foreachPartition算子后，可以获得以下的性能提升：

➢对于我们写的function函数，一次处理一整个分区的数据；  
➢对于一个分区内的数据，创建唯一的数据库连接；  
➢只需要向数据库发送一次SQL语句和多组参数；

在生产环境中，全部都会使用foreachPartition算子完成数据库操作。foreachPartition算子存在一个问题，与mapPartitions算子类似，如果一个分区的数据量特别大，可能会造成OOM，即内存溢出。

### 3. filter 与 coalesce 的配合使用 ###

在Spark任务中我们经常会使用filter算子完成RDD中数据的过滤，在任务初始阶段，从各个分区中加载到的数据量是相近的，但是一旦进过filter过滤后，每个分区的数据量有可能会存在较大差异，如图所示：

![在这里插入图片描述][0cc0d47ecfb84e98bc0d350d613f36e6.png]  
根据图中信息我们可以发现两个问题：

➢ 每个partition的数据量变小了，如果还按照之前与partition相等的task个数去处理当前数据，有点浪费task的计算资源；

➢ 每个partition的数据量不一样，会导致后面的每个task处理每个partition数据的时候，每个task要处理的数据量不同，这很有可能导致数据倾斜问题。

如上图所示，第二个分区的数据过滤后只剩100条，而第三个分区的数据过滤后剩下800条，在相同的处理逻辑下，第二个分区对应的task处理的数据量与第三个分区对应的task处理的数据量差距达到了8倍，这也会导致运行速度可能存在数倍的差距，这也就是数据倾斜问题。

针对上述的两个问题，我们分别进行分析：

➢针对第一个问题，既然分区的数据量变小了，我们希望可以对分区数据进行重新分配，比如将原来4个分区的数据转化到2个分区中，这样只需要用后面的两个task进行处理即可，避免了资源的浪费。

➢针对第二个问题，解决方法和第一个问题的解决方法非常相似，对分区数据重新分配，让每个partition中的数据量差不多，这就避免了数据倾斜问题。

那么具体应该如何实现上面的解决思路？我们需要 coalesce 算子。repartition 与 coalesce 都可以用来进行重分区，其中 repartition 只是 coalesce 接口中 shuffle为 true 的简易实现，coalesce 默认情况下不进行 shuffle，但是可以通过参数进行设置。假设我们希望将原本的分区个数 A 通过重新分区变为 B，那么有以下几种情况：

**➢A > B（多数分区合并为少数分区）**

A、A与B相差值不大此时使用coalesce即可，无需shuffle过程。

B、A与B相差值很大

此时可以使用coalesce并且不启用shuffle过程，但是会导致合并过程性能低下，所以推荐设置coalesce的第二个参数为true，即启动shuffle过程。

**➢A < B（少数分区分解为多数分区）**

此时使用 repartition 即可，如果使用 coalesce 需要将 shuffle 设置为 true，否则 coalesce 无效。我们可以在filter 操作之后，使用coalesce算子针对每个partition的数据量各不相同的情况，压缩partition的数量，而且让每个partition的数据量尽量均匀紧凑，以便于后面的task进行计算操作，在某种程度上能够在一定程度上提升性能。

注意：local模式是进程内模拟集群运行，已经对并行度和分区数量有了一定的内部优化，因此不用去设置并行度和分区数量。

### 4. repartition解决SparkSQL低并行度问题 ###

在第一节的常规性能调优中我们讲解了并行度的调节策略，但是，并行度的设置对于 Spark SQL 是不生效的，用户设置的并行度只对于 Spark SQL 以外的所有 Spark 的 stage 生效。

Spark SQL 的并行度不允许用户自己指定，Spark SQL自己会默认根据 hive 表对应的 HDFS 文件的 split 个数自动设置 Spark SQL 所在的那个 stage 的并行度，用户自己通 spark.default.parallelism 参数指定的并行度，只会在没 Spark SQL的stage中生效。

由于Spark SQL所在stage的并行度无法手动设置，如果数据量较大，并且此stage中后续的transformation操作有着复杂的业务逻辑，而Spark SQL自动设置的task数量很少，这就意味着每个task要处理为数不少的数据量，然后还要执行非常复杂的处理逻辑，这就可能表现为第一个有Spark SQL的 stage 速度很慢，而后续的没有 Spark SQL 的 stage 运行速度非常快。

为了解决Spark SQL无法设置并行度和task数量的问题，我们可以使用repartition算子。

![在这里插入图片描述][1d9b53fec427487594f6defffa215e0a.png]  
Spark SQL这一步的并行度和task数量肯定是没有办法去改变了，但是，对于Spark SQL查询出来的RDD，立即使用repartition算子，去重新进行分区，这样可以重新分区为多个partition，从repartition之后的RDD操作，由于不再设计Spark SQL，因此stage的并行度就会等于你手动设置的值，这样就避免了Spark SQL所在的stage只能用少量的task去处理大量数据并执行复杂的算法逻辑。

### 5. reduceByKey 预聚合 ###

reduceByKey相较于普通的shuffle操作一个显著的特点就是会进行map端的本地聚合，map端会先对本地的数据进行combine操作，然后将数据写入给下个stage的每个task创建的文件中，也就是在map端，对每一个key对应的value，执行reduceByKey算子函数。reduceByKey算子的执行过程如图所示：

![在这里插入图片描述][ce3545b466fc4a4e913470fbf7438c34.png]  
使用reduceByKey对性能的提升如下：

➢本地聚合后，在map端的数据量变少，减少了磁盘IO，也减少了对磁盘空间的占用；  
➢本地聚合后，下一个stage拉取的数据量变少，减少了网络传输的数据量；  
➢本地聚合后，在reduce端进行数据缓存的内存占用减少；  
➢本地聚合后，在reduce端进行聚合的数据量减少。

基于reduceByKey的本地聚合特征，我们应该考虑使用reduceByKey代替其他的shuffle算子，例如groupByKey。reduceByKey与groupByKey的运行原理如图所示：

![在这里插入图片描述][b03e0c8fd6b84fb1b13a47c63d4a4835.png]  
![在这里插入图片描述][869b4043f33c4c8282c6aa26283b9305.png]  
根据上图可知，groupByKey 不会进行 map 端的聚合，而是将所有 map 端的数据 shuffle 到 reduce 端，然后在 reduce 端进行数据的聚合操作。由于 reduceByKey 有 map 端聚合的特性，使得网络传输的数据量减小，因此效率要明显高于 groupByKey。

## 三、Shuffle调优 ##

### 1. 调节map端缓冲区大小 ###

在 Spark 任务运行过程中，如果 shuffle 的 map 端处理的数据量比较大，但是 map 端缓冲的大小是固定的，可能会出现 map 端缓冲数据频繁 spill 溢写到磁盘文件中的情况，使得性能非常低下，通过调节map 端缓冲的大小，可以避免频繁的磁盘IO操作，进而提升Spark任务的整体性能。

map 端缓冲的默认配置是 32KB，如果每个 task 处理 640KB 的数据，那么会发生 640/32 = 20 次溢写，如果每个 task 处理 64000KB的数据，机会发生64000/32=2000此溢写，这对于性能的影响是非常严重的。

map端缓冲的配置方法如代码清单所示：

val conf = new SparkConf().set("spark.shuffle.file.buffer", "64")

### 2. 调节reduce端拉取数据缓冲区大小 ###

Spark Shuffle过程中，shuffle reduce task的 buffer缓冲区大小决定了reduce task每次能够缓冲的数据量，也就是每次能够拉取的数据量，如果内存资源较为充足，适当增加拉取数据缓冲区的大小，可以减少拉取数据的次数，也就可以减少网络传输的次数，进而提升性能。

reduce 端数据拉取缓冲区的大小可以通过 spark.reducer.maxSizeInFlight 参数进行设置，默认为48MB，该参数的设置方法如代码清单所示：

val conf = new SparkConf().set("spark.reducer.maxSizeInFlight", "96")

### 3. 调节reduce端拉取数据重试次数 ###

Spark Shuffle 过程中，reduce task 拉取属于自己的数据时，如果因为网络异常等原因导致失败会自动进行重试。对于那些包含了特别耗时的 shuffle 操作的作业，建议增加重试最大次数 (比如60次)，以避免由于 JVM 的full gc或者网络不稳定等因素导致的数据拉取失败。在实践中发现，对于针对超大数据量（数十亿~上百亿）的shuffle过程，调节该参数可以大幅度提升稳定性。

reduce端拉取数据重试次数可以通过spark.shuffle.io.maxRetries参数进行设置，该参数就代表了可以重试的最大次数。如果在指定次数之内拉取还是没有成功，就可能会导致作业执行失败，默认为3，该参数的设置方法如代码清单所示：

val conf = new SparkConf().set("spark.shuffle.io.maxRetries", "6")

### 4. 调节reduce端拉取数据等待间隔 ###

Spark Shuffle过程中，reduce task拉取属于自己的数据时，如果因为网络异常等原因导致失败会自动进行重试，在一次失败后，会等待一定的时间间隔再进行重试，可以通过加大间隔时长 (比如60s)，以增加shuffle 操作的稳定性。reduce端拉取数据等待间隔可以通过 spark.shuffle.io.retryWait 参数进行设置，默认值为 5s，该参数的设置方法如代码清单所示：

val conf = new SparkConf().set("spark.shuffle.io.retryWait", "60s")

### 5. 调节SortShuffle排序操作阈值 ###

对于SortShuffleManager，如果shuffle reduce task的数量小于某一阈值则shuffle write过程中不会进行排序操作，而是直接按照未经优化的HashShuffleManager的方式去写数据，但是最后会将每个task产生的所有临时磁盘文件都合并成一个文件，并会创建单独的索引文件。

当你使用SortShuffleManager时，如果的确不需要排序操作，那么建议将这个参数调大一些，大于shuffle read task的数量，那么此时map-side就不会进行排序了，减少了排序的性能开销，但是这种方式下，依然会产生大量的磁盘文件，因此shuffle write性能有待提高。SortShuffleManager排序操作阈值的设置可以通过spark.shuffle.sort. bypassMergeThreshold这一参数进行设置，默认值为200，该参数的设置方法如代码清单所示：

val conf = new SparkConf().set("spark.shuffle.sort.bypassMergeThreshold", "400")

## 四、JVM调优 ##

对于JVM调优，首先应该明确，full gc/minor gc，都会导致JVM的工作线程停止工作，即stop the world。

### 1. 降低cache操作的内存占比 ###

A、静态内存管理机制根据Spark静态内存管理机制，堆内存被划分为了两块，Storage和Execution。  
Storage 主要用于缓存 RDD 数据和 broadcast 数据，Execution 主要用于缓存在shuffle过程中产生的中间数据，Storage占系统内存的60%，Execution占系统内存的20%，并且两者完全独立。在一般情况下，Storage的内存都提供给了cache操作，但是如果在某些情况下cache操作内存不是很紧张，而task的算子中创建的对象很多，Execution内存又相对较小，这回导致频繁的minor gc，甚至于频繁的full gc，进而导致Spark频繁的停止工作，性能影响会很大。

在Spark UI中可以查看每个stage的运行情况，包括每个task的运行时间、gc时间等等，如果发现gc太频繁，时间太长，就可以考虑调节Storage的内存占比，让task执行算子函数式，有更多的内存可以使用。Storage内存区域可以通过spark.storage.memoryFraction参数进行指定，默认为0.6，即60%，可以逐级向下递减，如代码清单所示：

val conf = new SparkConf().set("spark.storage.memoryFraction", "0.4")

B、统一内存管理机制根据Spark统一内存管理机制，堆内存被划分为了两块，Storage和Execution。Storage主要用于缓存数据，Execution主要用于缓存在shuffle过程中产生的中间数据，两者所组成的内存部分称为统一内存，Storage和Execution各占统一内存的50%，由于动态占用机制的实现，shuffle过程需要的内存过大时，会自动占用Storage的内存区域，因此无需手动进行调节。

### 2. 调节Executor堆外内存 ###

Executor的堆外内存主要用于程序的共享库、PermSpace、线程Stack和一些Memory mapping等, 或者类C方式allocate object。有时，如果你的Spark作业处理的数据量非常大，达到几亿的数据量，此时运行Spark作业会时不时地报错，例如shuffle output file cannot find，executor lost，task lost，out of memory等，这可能是Executor的堆外内存不太够用，导致Executor在运行的过程中内存溢出。

stage的task在运行的时候，可能要从一些Executor中去拉取shuffle map output文件，但是Executor可能已经由于内存溢出挂掉了，其关联的BlockManager也没有了，这就可能会报出shuffle output file cannot find，executor lost，task lost，out of memory等错误，此时，就可以考虑调节一下Executor的堆外内存，也就可以避免报错，与此同时，堆外内存调节的比较大的时候，对于性能来讲，也会带来一定的提升。

默认情况下，Executor堆外内存上限大概为300多MB，在实际的生产环境下，对海量数据进行处理的时候，这里都会出现问题，导致Spark作业反复崩溃，无法运行，此时就会去调节这个参数，到至少1G，甚至于2G、4G。

Executor堆外内存的配置需要在spark-submit脚本里配置，如代码清单所示：

--conf spark.yarn.executor.memoryOverhead=2048

以上参数配置完成后，会避免掉某些JVM OOM的异常问题，同时，可以提升整体 Spark 作业的性能。

### 3. 调节连接等待时长 ###

在Spark作业运行过程中，Executor优先从自己本地关联的BlockManager中获取某份数据，如果本地BlockManager没有的话，会通过TransferService远程连接其他节点上Executor的BlockManager来获取数据。

如果task在运行过程中创建大量对象或者创建的对象较大，会占用大量的内存，这回导致频繁的垃圾回收，但是垃圾回收会导致工作现场全部停止，也就是说，垃圾回收一旦执行，Spark的Executor进程就会停止工作，无法提供相应，此时，由于没有响应，无法建立网络连接，会导致网络连接超时。

在生产环境下，有时会遇到file not found、file lost这类错误，在这种情况下，很有可能是 Executor 的BlockManager在拉取数据的时候，无法建立连接，然后超过默认的连接等待时长60s后，宣告数据拉取失败，如果反复尝试都拉取不到数据，可能会导致Spark作业的崩溃。这种情况也可能会导致DAGScheduler反复提交几次stage，TaskScheduler返回提交几次task，大大延长了我们的Spark作业的运行时间。

此时，可以考虑调节连接的超时时长，连接等待时长需要在 spark-submit 脚本中进行设置，设置方式如代码清单所示：`--conf spark.core.connection.ack.wait.timeout=300` 调节连接等待时长后，通常可以避免部分的XX文件拉取失败、XX文件 lost 等报错。

[a7c9bd51118848e882a7bfedc123b377.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/0eb00b393407445ea371e7aaaeae2846.png
[7e8740fc29424e5d8859b7249ccbd3f9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/4a60fb5737de413bafcaef75d8cea04f.png
[5276900395354843ad6d341a3b8cc3c0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/c1a373f4c2bd44b3a42187fbd348f5ff.png
[9b60409a3345445492645336fcbb5bdc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/09f2e2ea073a4d039fee96ca7aa2c3d6.png
[5fa3d8d700e44a28ab16ca6d02004b0e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/1c6f7447bd724243898609f6ef3ab667.png
[fd61c26f5fa74c2d8f7cde5829162f63.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/ceed25e765a4468fba91c109dfff738a.png
[0cc0d47ecfb84e98bc0d350d613f36e6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/02aa0b97a783445cb3caa4e891561bc8.png
[1d9b53fec427487594f6defffa215e0a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/40227631b3b34586a6be9b24c78161ec.png
[ce3545b466fc4a4e913470fbf7438c34.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/95c4a71254e64bf1b9d7c6a045289fc9.png
[b03e0c8fd6b84fb1b13a47c63d4a4835.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/04df8da0e51e482b945c67a76a18d847.png
[869b4043f33c4c8282c6aa26283b9305.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/06/67e435bdeb86431d99650a8020f8bf45.png