Spark性能优化

末蓝、 2023-07-21 08:39 10阅读 0赞

## 1、资源参数调优 ##

### 1.1 运行时架构 ###

![E5_9B_BE_E7_89_87-6.png][]

#### 1.1.1 **Client** ： ####

客户端进程，负责提交作业

####  **1.1.2 Driver/SC ：** ####

运行应用程序/业务代码的main()函数并且创建SparkContext，其中创建SparkContext的目的是为了准备Spark应用程序的运行环境。在Spark中由SparkContext负责和ClusterManager/ResourceManager通信，进行资源的申请、任务的分配和监控等；当Executor部分运行完毕后，Driver负责将SparkContext关闭。通常用SparkContext代表Drive

*  SparkContext：整个应用程序的上下文，控制应用的生命周期
 *  DAGScheduler：实现将Spark作业分解成一到多个Stage，每个Stage根据RDD的Partition个数决定Task的个数，然后生成相应的Task set放到TaskScheduler中
 *  TaskScheduler：分配Task到Executor上执行，并维护Task的运行状态

#### 1.1.3 Executor ####

应用程序Application运行在Worker节点上的一个进程，该进程负责运行Task，并且负责将数据存在内存或者磁盘上。

在Spark on Yarn模式下，其进程名称为CoarseGrainedExecutorBackend，负责将Task包装成taskRunner，并从线程池中抽取出一个空闲线程运行Task。每个CoarseGrainedExecutorBackend能并行运行Task的数量就取决于分配给它的CPU的个数

#### 1.1.4 Job ####

一个job包含多个RDD及作用于相应RDD上的各种Operation。

每执行一个action算子（`foreach, count, collect, take, saveAsTextFile`）就会生成一个 job

#### 1.1.5 Stage ####

每个Job会被拆分很多组Task，每组Task被称为Stage，亦称TaskSet。

一个作业job分为多个阶段stages（shuffle，串行）

一个stage包含一系列的tasks（并行）

#### 1.1.6 Task ####

被送往各个Executor上的执行的内容，task之间无状态传递，可以并行执行

### 1.2 运行流程 ###

1.  client向YARN的ResourceManager/RM申请启动ApplicationMaster/AM（单个应用程序/作业的资源管理和任务监控）
2.  RM收到请求后，在集群中选择一个NodeManager，为该应用程序分配第一个Container，spark在此启动其AM，其中AM进行SparkContext/SC/Driver初始化启动并创建RDD Object、DAGScheduler、TASKScheduler
3.  SC根据RDD的依赖关系构建DAG图，并将DAG提交给DAGScheduler解析为**stage**。Stages以TaskSet的形式提交给TaskScheduler，TaskScheduler维护所有TaskSet，当Executor向Driver发送心跳时，TaskScheduler会根据其资源剩余情况分配相应的Task，另外TaskScheduler还维护着所有Task的运行状态，重试失败了的Task
4.  AM向RM申请container资源，资源到位后便与NodeManager通信，要求它在获得的Container中(executor)启动CoarseGrainedExecutorBackend，CoarseGrainedExecutorBackend启动后会向AM中的SC注册并申请Task
5.  AM中的SC分配Task给CoarseGrainedExecutorBackend/executor执行，CoarseGrainedExecutorBackend运行Task并向AM汇报运行的状态和进度，以便让AM随时掌握各个task的运行状态，从而可以在任务失败时重新启动任务或者推测执行
6.  应用程序运行完成后，AM向RM申请注销并关闭自己

### 1.3 调优 ###

1.  executor配置
    
     *  spark.executor.memory
     *  spark.executor.instances
     *  spark.executor.cores
2.  driver配置
    
     *  spark.driver.memory（如果没有collect操作，一般不需要很大，1~4g即可）
     *  spark.driver.cores
3.  并行度
    
     *  spark.default.parallelism (used for RDD API)
     *  spark.sql.shuffle.partitions (usef for DataFrame/DataSet API)
4.  网络超时
    
     *  spark.network.timeout (所有网络交互的默认超时)
5.  数据本地化
    
     *  spark.locality.wait
6.  JVM/gc配置
    
     *  spark.executor.extraJavaOptions
     *  spark.driver.extraJavaOptions

**生产配置**

1 /usr/local/spark/bin/spark-submit \
     2 --class com.duoduo.spark.WordCount \
     3 --num-executors 80 \
     4 --driver-memory 6g \
     5 --executor-memory 6g \
     6 --executor-cores 3 \
     7 --master yarn \
     8 --deploy-mode cluster \
     9 --queue root.default \
    10 --conf spark.executor.memoryOverhead=2048 \
    11 --conf spark.core.connection.ack.wait.timeout=300 \
    12 /usr/local/spark/spark.jar

1 executor-cores —— 每个executor使用的内核数，默认为1，官方建议2-5个
    2 num-executors —— 启动executors的数量，默认为2
    3 executor-memory —— executor内存大小，默认1G
    4 driver-cores —— driver使用内核数，默认为1
    5 driver-memory —— driver内存大小，默认512M

## 2、开发调优 ##

### 2.1 避免创建重复的RDD ###

![E5_9B_BE_E7_89_87-8.png][]

修改后

![E5_9B_BE_E7_89_87-9.png][]

### 2.2 尽可能复用同一个RDD，持久化 ###

但是如果rdd的lineage太长，最好checkpoint下来，避免长重建

持久化级别： （SER，MEM，DISK，\_N ）

### 2.3 尽量避免使用shuffle类算子 ###

*  shuffle算子如distinct（实际调用reduceByKey）、reduceByKey、aggregateByKey、sortByKey、groupByKey、join、cogroup、repartition等，入参中会有一个并行度参数numPartitions
 *  shuffle过程中，各个节点上的相同key都会先写入本地磁盘文件中，然后其他节点需要通过网络传输拉取各个节点上的磁盘文件中的相同key

### 2.4 使用map-side预聚合的shuffle操作 ###

reduceByKey(combiner)，groupByKey(没有combiner)

![E5_9B_BE_E7_89_87-10.png][]![E5_9B_BE_E7_89_87-11.png][]

### 2.5 使用高性能的算子 ###

#### **1. 减少小文件数量**  ####

#### **2. 特别是在写DB的时候，避免每条写记录都new一个connection；推荐是每个partition new一个connection；更好的是new connection池，每个partition从中取即可，减少partitionNum个new的消耗**  ####

1 dstream.foreachRDD { rdd =>
    2   rdd.foreachPartition { partitionOfRecords =>
    3     // ConnectionPool is a static, lazily initialized pool of connections
    4     val connection = ConnectionPool.getConnection()
    5     partitionOfRecords.foreach(record => connection.send(record))
    6     ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
    7   }
    8 }

#### **3. 使用reduceByKey/aggregateByKey（有预聚合）替代groupByKey**  ####

![E5_9B_BE_E7_89_87-12.png][]![E5_9B_BE_E7_89_87-13.png][]

#### **4.mapPartition代替普通的map** ####

#### **5.foreachPartition代替foreach** ####

使用了foreachPartition算子后，可以获得以下的性能提升：

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

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

#### **6. 使用filter之后进行coalesce操作** ####

![E5_9B_BE_E7_89_87-14.png][]

repartition与coalesce都可以用来进行重分区，其中repartition只是coalesce接口中shuffle为true的简易实现，coalesce默认情况下不进行shuffle，但是可以通过参数进行设置。

假设我们希望将原本的分区个数A通过重新分区变为B，那么有以下几种情况： 1. A > B（多数分区合并为少数分区）

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

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

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

此时使用repartition即可，如果使用coalesce需要将shuffle设置为true，否则coalesce无效。

总结: 我们可以在filter操作之后，使用coalesce算子针对每个partition的数据量各不相同的情况，压缩partition的数量，而且让每个partition的数据量尽量均匀紧凑，以便于后面的task进行计算操作，在某种程度上能够在一定程度上提升性能。

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

#### **7. 使用repartitionAndSortWithinPartitions替代repartition与sort类操作**  ####

####  **8. reparation解决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算子。

![E5_9B_BE_E7_89_87-15.png][]

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

### 2.6 广播大变量 ###

默认情况下，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中获取变量。

### 2.7 使用kryo序列化 ###

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

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

1 public class MyKryoRegistrator implements KryoRegistrator{
     2   @Override
     3   public void registerClasses(Kryo kryo){
     4     kryo.register(StartupReportLogs.class);
     5   }
     6 }
     7 //创建SparkConf对象
     8 val conf = new SparkConf().setMaster(…).setAppName(…)
     9 //使用Kryo序列化库，如果要使用Java序列化库，需要把该行屏蔽掉
    10 conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");  
    11 //在Kryo序列化库中注册自定义的类集合，如果要使用Java序列化库，需要把该行屏蔽掉
    12 conf.set("spark.kryo.registrator", "com.duoduo.MyKryoRegistrator");

### 2.8 调节本地化等待时间 ###

Spark 作业运行过程中，Driver 会对每一个 stage 的 task 进行分配。根据 Spark 的 task 分配算法，Spark希望task能够运行在它要计算的数据所在的节点（数据本地化思想），这样就可以避免数据的网络传输。

通常来说，task可能不会被分配到它处理的数据所在的节点，因为这些节点可用的资源可能已经用尽，此时，Spark会等待一段时间，默认3s

如果等待指定时间后仍然无法在指定节点运行，那么会自动降级，尝试将task分配到比较差的本地化级别所对应的节点上，比如将task分配到离它要计算的数据比较近的一个节点，然后进行计算，如果当前级别仍然不行，那么继续降级。

当task要处理的数据不在task所在节点上时，会发生数据的传输。

task会通过所在节点的BlockManager获取数据，BlockManager发现数据不在本地时，会通过网络传输组件从数据所在节点的BlockManager处获取数据。

大量的网络传输会严重影响性能，因此，我们希望通过调节本地化等待时长，如果在等待时长这段时间内，目标节点处理完成了一部分task，那么当前的task将有机会得到执行，这样就能够改善Spark作业的整体性能。

**Spark本地化等级**

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

在Spark项目开发阶段，可以使用client模式对程序进行测试，此时，可以在本地看到比较全的日志信息，日志信息中有明确的task数据本地化的级别，如果大部分都是PROCESS\_LOCAL，那么就无需进行调节，但是如果发现很多的级别都是NODE\_LOCAL、ANY，那么需要对本地化的等待时长进行调节，通过延长本地化等待时长，看看task的本地化级别有没有提升，并观察Spark作业的运行时间有没有缩短。

注意，过犹不及，不要将本地化等待时长延长地过长，导致因为大量的等待时长，使得Spark作业的运行时间反而增加了。

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

## 3、Shuffle调优 ##

### 3.1 shuffle原理 ###

*  Spark在DAG阶段以宽依赖shuffle为界，划分stage，上游stage做map task，每个map task将计算结果数据分成多份，每一份对应到下游stage的每个partition中，并将其临时写到磁盘，该过程叫做shuffle write

*  下游stage做reduce task，每个reduce task通过网络拉取上游stage中所有map task的指定分区结果数据，该过程叫做shuffle read，最后完成reduce的业务逻辑

*  下图中，上游stage有3个map task，下游stage有4个reduce task，那么这3个map task中**每个map task都会产生4份数据**。而4个reduce task中的每个reduce task都会拉取上游3个map task对应的那份数据

![E5_9B_BE_E7_89_87-16.png][]

### 3.2 Shuffle演变 ###

*  **<0.8** hashBasedShuffle
    
     *  每个map端的task为每个reduce端的partition/task生成一个文件，通常会产生大量的文件，伴随大量的随机磁盘IO操作与大量的内存开销`M*R`
 *  **0.8.1** 引入文件合并File Consolidation机制
    
     *  每个executor为每个reduce端的partition生成一个文件`E*R`
 *  **0.9** 引入External AppendOnlyMap
    
     *  combine时可以将数据spill到磁盘，然后通过堆排序merge
 *  **1.1** 引入sortBasedShuffle
    
     *  每个map task不会为每个reducer task生成一个单独的文件，而是会将所有的结果写到一个文件里，同时会生成一个index文件，reducer可以通过这个index文件取得它需要处理的数据`M`
 *  **1.4** 引入Tungsten-Sort Based Shuffle
    
     *  亦称unsafeShuffle，将数据记录用序列化的二进制方式存储，把排序转化成指针数组的排序，引入堆外内存空间和新的内存管理模型
 *  **1.6** Tungsten-sort并入Sort Based Shuffle
    
     *  由SortShuffleManager自动判断选择最佳Shuffle方式，如果检测到满足Tungsten-sort条件会自动采用Tungsten-sort Based Shuffle，否则采用Sort Based Shuffle
 *  **2.0** hashBasedShuffle退出历史舞台
    
     *  从此Spark只有sortBasedShuffle

### 3.3 调优 ###

shuffle是一个涉及到CPU（序列化反序列化）、网络IO（跨节点数据传输）以及磁盘IO（shuffle中间结果落盘）的操作。

所以用户在编写Spark应用程序的过程中应当尽可能避免shuffle算子和考虑shuffle相关的优化，提升spark应用程序的性能。

要减少shuffle的开销，主要有两个思路，

*  减少shuffle次数，尽量不改变key，把数据处理在local完成
 *  减少shuffle的数据规模

**（1）先去重，再合并**

`A.union(B).distinct()`  vs  `A.distinct().union(B.distinct()).distinct()`

**（2） 用broadcast + filter来代替join**

**（3）*调节*map*端缓冲区大小***

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

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

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

**（4）调节reduce端拉取数据缓冲区大小**

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

reduce端数据拉取缓冲区的大小可以通过`spark.reducer.maxSizeInFligh`t参数进行设置**，默认为48MB**

**（5）调节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")

**（6）调节reduce端拉取数据等待间隔**

Spark Shuffle 过程中，reduce task 拉取属于自己的数据时，如果因为网络异常等原因导致失败会自动进行重试，在一次失败后，会等待一定的时间间隔再进行重试，可以通过加大间隔时长（比如60s），以增加shuffle操作的稳定性。

reduce端拉取数据等待间隔可以通过`spark.shuffle.io.retryWait`参数进行设置，**默认值为5s**

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

**（7）调节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")

## 4、Spark的Join类型 ##

Shuffled Hash Join  
Sort Merge Join  
Broadcast Join

![E5_9B_BE_E7_89_87-17.png][]

## 5、JVM优化 ##

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

### 5.1 降低cache操作的内存占比 ###

**静态内存管理机制**

根据 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")

**同一内存管理机制**

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

### 5.2 调节Executor堆外内存 ###

Executor 的堆外内存主要用于程序的共享库、Perm Space、 线程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.executor.memoryOverhead=2048`

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

### 5.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等报错。

## 6、其他优化 ##

使用DataFrame/DataSet

*  spark sql 的catalyst优化器
 *  堆外内存（有了Tungsten后，感觉off-head没有那么明显的性能提升了）

![E5_9B_BE_E7_89_87-18.png][]

<table> 
 <tbody> 
  <tr> 
   <td><strong>Type<br></strong></td> 
   <td><strong>RDD<br></strong></td> 
   <td><strong>DataFrame</strong></td> 
   <td><strong>DataSet</strong></td> 
  </tr> 
  <tr> 
   <td>definition</td> 
   <td>RDD是分布式的Java对象的集合</td> 
   <td>DataFrame是分布式的Row对象的集合</td> 
   <td>DataSet是分布式的Java对象的集合<br>ds = df.as[ElementType]<br>df = Dataset[Row]</td> 
  </tr> 
  <tr> 
   <td>pros</td> 
   <td>* 编译时类型安全<br>* 面向对象的编程风格</td> 
   <td>* 引入schema结构信息<br>* 减少数据读取，优化执行计划，如filter下推，剪裁<br>* off-heap堆外存储</td> 
   <td>* Encoder序列化<br>* 支持结构与非结构化数据<br>* 和rdd一样，支持自定义对象存储<br>* 和dataframe一样，支持结构化数据的sql查询<br>* 采用堆外内存存储，gc友好<br>* 类型转化安全，代码有好</td> 
  </tr> 
  <tr> 
   <td>cons</td> 
   <td>* 对于结构化数据不友好<br>* 默认采用的是java序列化方式，序列化结果比较大，而且数据存储在java堆内存中，导致gc比较频繁</td> 
   <td><br>* rdd内部数据直接以java对象存储，dataframe内存存储的是Row对象而不能是自定义对象<br>* 编译时不能类型转化安全检查，运行时才能确定是否有问题</td> 
   <td>* 可能需要额外定义Encoder</td> 
  </tr> 
 </tbody> 
</table>

[E5_9B_BE_E7_89_87-6.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-6.png
[E5_9B_BE_E7_89_87-8.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-8.png
[E5_9B_BE_E7_89_87-9.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-9.png
[E5_9B_BE_E7_89_87-10.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-10.png
[E5_9B_BE_E7_89_87-11.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-11.png
[E5_9B_BE_E7_89_87-12.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-12.png
[E5_9B_BE_E7_89_87-13.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-13.png
[E5_9B_BE_E7_89_87-14.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-14.png
[E5_9B_BE_E7_89_87-15.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-15.png
[E5_9B_BE_E7_89_87-16.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-16.png
[E5_9B_BE_E7_89_87-17.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-17.png
[E5_9B_BE_E7_89_87-18.png]: http://47.112.223.94/wp-content/uploads/2020/04/%E5%9B%BE%E7%89%87-18.png