大数据开发-Spark调优经常使用手段

Spark调优

spark调优常见手段，在生产中经常会遇到各类各样的问题，有事前缘由，有事中缘由，也有不规范缘由，spark调优总结下来能够从下面几个点来调优。

1. 分配更多的资源

分配更多的资源：
  它是性能优化调优的王道，就是增长和分配更多的资源，这对于性能和速度上的提高是显而易见的，
  基本上，在必定范围以内，增长资源与性能的提高，是成正比的；写完了一个复杂的spark做业以后，进行性能调优的时候，首先第一步，就是要来调节最优的资源配置；
  在这个基础之上，若是说你的spark做业，可以分配的资源达到了你的能力范围的顶端以后，没法再分配更多的资源了，公司资源有限；那么才是考虑去作后面的这些性能调优的点。

相关问题：
（1）分配哪些资源？
（2）在哪里能够设置这些资源？
（3）剖析为何分配这些资源以后，性能能够获得提高？

1.1 分配哪些资源

executor-memory、executor-cores、driver-memory

1.2 在哪里能够设置这些资源

在实际的生产环境中，提交spark任务时，使用spark-submit shell脚本，在里面调整对应的参数。
   
 提交任务的脚本:
 spark-submit \
 --master spark://node1:7077 \
 --class com.hoult.WordCount \
 --num-executors 3 \    配置executor的数量
 --driver-memory 1g \   配置driver的内存（影响不大）
 --executor-memory 1g \ 配置每个executor的内存大小
 --executor-cores 3 \   配置每个executor的cpu个数
 /export/servers/wordcount.jar

1.2 参数调节到多大，算是最大

• Standalone模式

先计算出公司spark集群上的全部资源 每台节点的内存大小和cpu核数，
   好比：一共有20台worker节点，每台节点8g内存，10个cpu。
   实际任务在给定资源的时候，能够给20个executor、每一个executor的内存8g、每一个executor的使用的cpu个数10。

• Yarn模式

先计算出yarn集群的全部大小，好比一共500g内存，100个cpu；
   这个时候能够分配的最大资源，好比给定50个executor、每一个executor的内存大小10g,每一个executor使用的cpu个数为2。

• 使用原则

在资源比较充足的状况下，尽量的使用更多的计算资源，尽可能去调节到最大的大小

1.3 为何调大资源之后性能能够提高

--executor-memory

--total-executor-cores

2. 提升并行度

2.1 Spark的并行度指的是什么

spark做业中，各个stage的task的数量，也就表明了spark做业在各个阶段stage的并行度！
    当分配完所能分配的最大资源了，而后对应资源去调节程序的并行度，若是并行度没有与资源相匹配，那么致使你分配下去的资源都浪费掉了。同时并行运行，还可让每一个task要处理的数量变少（很简单的原理。合理设置并行度，能够充分利用集群资源，减小每一个task处理数据量，而增长性能加快运行速度。）

2.2 如何提升并行度

2.2.1 能够设置task的数量

至少设置成与spark Application 的总cpu core 数量相同。
最理想状况，150个core，分配150task，一块儿运行，差很少同一时间运行完毕
官方推荐，task数量，设置成spark Application 总cpu core数量的2~3倍 。
  
好比150个cpu core ，基本设置task数量为300~500. 与理想状况不一样的，有些task会运行快一点，好比50s就完了，有些task 可能会慢一点，要一分半才运行完，因此若是你的task数量，恰好设置的跟cpu core 数量相同，可能会致使资源的浪费。
  由于好比150个task中10个先运行完了，剩余140个还在运行，可是这个时候，就有10个cpu core空闲出来了，致使浪费。若是设置2~3倍，那么一个task运行完之后，另一个task立刻补上来，尽可能让cpu core不要空闲。同时尽可能提高spark运行效率和速度。提高性能。

2.2.2 如何设置task数量来提升并行度

设置参数spark.default.parallelism
   默认是没有值的，若是设置了值为10，它会在shuffle的过程才会起做用。
   好比 val rdd2 = rdd1.reduceByKey(_+_) 
   此时rdd2的分区数就是10
   
能够经过在构建SparkConf对象的时候设置，例如：
   new SparkConf().set("spark.defalut.parallelism","500")

2.2.3 给RDD从新设置partition的数量

使用rdd.repartition 来从新分区，该方法会生成一个新的rdd，使其分区数变大。
此时因为一个partition对应一个task，那么对应的task个数越多，经过这种方式也能够提升并行度。

2.2.4 提升sparksql运行的task数量

http://spark.apache.org/docs/2.3.3/sql-programming-guide.html

经过设置参数 spark.sql.shuffle.partitions=500  默认为200；
能够适当增大，来提升并行度。 好比设置为 spark.sql.shuffle.partitions=500

专门针对sparkSQL来设置的

3. RDD的重用和持久化

3.1 实际开发遇到的状况说明

如上图所示的计算逻辑：
（1）当第一次使用rdd2作相应的算子操做获得rdd3的时候，就会从rdd1开始计算，先读取HDFS上的文件，而后对rdd1作对应的算子操做获得rdd2,再由rdd2计算以后获得rdd3。一样为了计算获得rdd4，前面的逻辑会被从新计算。

（3）默认状况下屡次对一个rdd执行算子操做，去获取不一样的rdd，都会对这个rdd及以前的父rdd所有从新计算一次。
这种状况在实际开发代码的时候会常常遇到，可是咱们必定要避免一个rdd重复计算屡次，不然会致使性能急剧下降。

总结：能够把屡次使用到的rdd，也就是公共rdd进行持久化，避免后续须要，再次从新计算，提高效率。

3.2 如何对rdd进行持久化

• 能够调用rdd的cache或者persist方法。

（1）cache方法默认是把数据持久化到内存中 ，例如：rdd.cache ，其本质仍是调用了persist方法
（2）persist方法中有丰富的缓存级别，这些缓存级别都定义在StorageLevel这个object中，能够结合实际的应用场景合理的设置缓存级别。例如： rdd.persist(StorageLevel.MEMORY_ONLY),这是cache方法的实现。

3.3 rdd持久化的时能够采用序列化

（1）若是正常将数据持久化在内存中，那么可能会致使内存的占用过大，这样的话，也许会致使OOM内存溢出。
（2）当纯内存没法支撑公共RDD数据彻底存放的时候，就优先考虑使用序列化的方式在纯内存中存储。将RDD的每一个partition的数据，序列化成一个字节数组；序列化后，大大减小内存的空间占用。
（3）序列化的方式，惟一的缺点就是，在获取数据的时候，须要反序列化。可是能够减小占用的空间和便于网络传输
（4）若是序列化纯内存方式，仍是致使OOM，内存溢出；就只能考虑磁盘的方式，内存+磁盘的普通方式（无序列化）。
（5）为了数据的高可靠性，并且内存充足，可使用双副本机制，进行持久化
  持久化的双副本机制，持久化后的一个副本，由于机器宕机了，副本丢了，就仍是得从新计算一次；
  持久化的每一个数据单元，存储一份副本，放在其余节点上面，从而进行容错；
  一个副本丢了，不用从新计算，还可使用另一份副本。这种方式，仅仅针对你的内存资源极度充足。
   好比: StorageLevel.MEMORY_ONLY_2

4. 广播变量的使用

4.1 场景描述

在实际工做中可能会遇到这样的状况，因为要处理的数据量很是大，这个时候可能会在一个stage中出现大量的task，好比有1000个task，这些task都须要一份相同的数据来处理业务，这份数据的大小为100M，该数据会拷贝1000份副本，经过网络传输到各个task中去，给task使用。这里会涉及大量的网络传输开销，同时至少须要的内存为1000*100M=100G，这个内存开销是很是大的。没必要要的内存的消耗和占用，就致使了你在进行RDD持久化到内存，也许就无法彻底在内存中放下；就只能写入磁盘，最后致使后续的操做在磁盘IO上消耗性能；这对于spark任务处理来讲就是一场灾难。

    因为内存开销比较大，task在建立对象的时候，可能会出现堆内存放不下全部对象，就会致使频繁的垃圾回收器的回收GC。GC的时候必定是会致使工做线程中止，也就是致使Spark暂停工做那么一点时间。频繁GC的话，对Spark做业的运行的速度会有至关可观的影响。

4.2 广播变量引入

Spark中分布式执行的代码须要传递到各个executor的task上运行。对于一些只读、固定的数据,每次都须要Driver广播到各个Task上，这样效率低下。广播变量容许将变量只广播给各个executor。该executor上的各个task再从所在节点的BlockManager(负责管理某个executor对应的内存和磁盘上的数据)获取变量，而不是从Driver获取变量，从而提高了效率。

广播变量，初始的时候，就在Drvier上有一份副本。经过在Driver把共享数据转换成广播变量。

  task在运行的时候，想要使用广播变量中的数据，此时首先会在本身本地的Executor对应的BlockManager中，尝试获取变量副本；若是本地没有，那么就从Driver远程拉取广播变量副本，并保存在本地的BlockManager中；
  
  此后这个executor上的task，都会直接使用本地的BlockManager中的副本。那么这个时候全部该executor中的task都会使用这个广播变量的副本。也就是说一个executor只须要在第一个task启动时，得到一份广播变量数据，以后的task都从本节点的BlockManager中获取相关数据。

  executor的BlockManager除了从driver上拉取，也可能从其余节点的BlockManager上拉取变量副本，网络距离越近越好。

4.3 使用广播变量后的性能分析

好比一个任务须要50个executor，1000个task，共享数据为100M。
(1)在不使用广播变量的状况下，1000个task，就须要该共享数据的1000个副本，也就是说有1000份数须要大量的网络传输和内存开销存储。耗费的内存大小1000*100=100G.

(2)使用了广播变量后，50个executor就只须要50个副本数据，并且不必定都是从Driver传输到每一个节点，还多是就近从最近的节点的executor的blockmanager上拉取广播变量副本，网络传输速度大大增长；内存开销 50*100M=5G

总结：
  不使用广播变量的内存开销为100G，使用后的内存开销5G，这里就相差了20倍左右的网络传输性能损耗和内存开销，使用广播变量后对于性能的提高和影响，仍是很可观的。
  
  广播变量的使用不必定会对性能产生决定性的做用。好比运行30分钟的spark做业，可能作了广播变量之后，速度快了2分钟，或者5分钟。可是一点一滴的调优，聚沙成塔。最后仍是会有效果的。

4.4 广播变量使用注意事项

（1）能不能将一个RDD使用广播变量广播出去？

       不能，由于RDD是不存储数据的。能够将RDD的结果广播出去。

（2）广播变量只能在Driver端定义，不能在Executor端定义。

（3）在Driver端能够修改广播变量的值，在Executor端没法修改广播变量的值。

（4）若是executor端用到了Driver的变量，若是不使用广播变量在Executor有多少task就有多少Driver端的变量副本。

（5）若是Executor端用到了Driver的变量，若是使用广播变量在每一个Executor中只有一份Driver端的变量副本。

4.5 如何使用广播变量

• 例如

(1) 经过sparkContext的broadcast方法把数据转换成广播变量，类型为Broadcast，
  val broadcastArray: Broadcast[Array[Int]] = sc.broadcast(Array(1,2,3,4,5,6))
  
(2) 而后executor上的BlockManager就能够拉取该广播变量的副本获取具体的数据。
    获取广播变量中的值能够经过调用其value方法
   val array: Array[Int] = broadcastArray.value

5. 尽可能避免使用shuffle类算子

5.1 shuffle描述

spark中的shuffle涉及到数据要进行大量的网络传输，下游阶段的task任务须要经过网络拉取上阶段task的输出数据，shuffle过程，简单来讲，就是将分布在集群中多个节点上的同一个key，拉取到同一个节点上，进行聚合或join等操做。好比reduceByKey、join等算子，都会触发shuffle操做。
  
  若是有可能的话，要尽可能避免使用shuffle类算子。
  由于Spark做业运行过程当中，最消耗性能的地方就是shuffle过程。

5.2 哪些算子操做会产生shuffle

spark程序在开发的过程当中使用reduceByKey、join、distinct、repartition等算子操做，这里都会产生shuffle，因为shuffle这一块是很是耗费性能的，实际开发中尽可能使用map类的非shuffle算子。这样的话，没有shuffle操做或者仅有较少shuffle操做的Spark做业，能够大大减小性能开销。

5.3 如何避免产生shuffle

• 小案例

//错误的作法：
// 传统的join操做会致使shuffle操做。
// 由于两个RDD中，相同的key都须要经过网络拉取到一个节点上，由一个task进行join操做。
val rdd3 = rdd1.join(rdd2)
    
//正确的作法：
// Broadcast+map的join操做，不会致使shuffle操做。
// 使用Broadcast将一个数据量较小的RDD做为广播变量。
val rdd2Data = rdd2.collect()
val rdd2DataBroadcast = sc.broadcast(rdd2Data)

// 在rdd1.map算子中，能够从rdd2DataBroadcast中，获取rdd2的全部数据。
// 而后进行遍历，若是发现rdd2中某条数据的key与rdd1的当前数据的key是相同的，那么就断定能够进行join。
// 此时就能够根据本身须要的方式，将rdd1当前数据与rdd2中能够链接的数据，拼接在一块儿（String或Tuple）。
val rdd3 = rdd1.map(rdd2DataBroadcast...)

// 注意，以上操做，建议仅仅在rdd2的数据量比较少（好比几百M，或者一两G）的状况下使用。
// 由于每一个Executor的内存中，都会驻留一份rdd2的全量数据。

5.4 使用map-side预聚合的shuffle操做

• map-side预聚合

若是由于业务须要，必定要使用shuffle操做，没法用map类的算子来替代，那么尽可能使用能够map-side预聚合的算子。

  所谓的map-side预聚合，说的是在每一个节点本地对相同的key进行一次聚合操做，相似于MapReduce中的本地combiner。
  map-side预聚合以后，每一个节点本地就只会有一条相同的key，由于多条相同的key都被聚合起来了。其余节点在拉取全部节点上的相同key时，就会大大减小须要拉取的数据数量，从而也就减小了磁盘IO以及网络传输开销。
  一般来讲，在可能的状况下，建议使用reduceByKey或者aggregateByKey算子来替代掉groupByKey算子。由于reduceByKey和aggregateByKey算子都会使用用户自定义的函数对每一个节点本地的相同key进行预聚合。
  而groupByKey算子是不会进行预聚合的，全量的数据会在集群的各个节点之间分发和传输，性能相对来讲比较差。
  
  好比以下两幅图，就是典型的例子，分别基于reduceByKey和groupByKey进行单词计数。其中第一张图是groupByKey的原理图，能够看到，没有进行任何本地聚合时，全部数据都会在集群节点之间传输；第二张图是reduceByKey的原理图，能够看到，每一个节点本地的相同key数据，都进行了预聚合，而后才传输到其余节点上进行全局聚合。

• groupByKey进行单词计数原理

• reduceByKey单词计数原理

6. 使用高性能的算子

6.1 使用reduceByKey/aggregateByKey替代groupByKey

• reduceByKey/aggregateByKey 能够进行预聚合操做，减小数据的传输量，提高性能

• groupByKey 不会进行预聚合操做，进行数据的全量拉取，性能比较低

6.2 使用mapPartitions替代普通map

mapPartitions类的算子，一次函数调用会处理一个partition全部的数据，而不是一次函数调用处理一条，性能相对来讲会高一些。
  可是有的时候，使用mapPartitions会出现OOM（内存溢出）的问题。由于单次函数调用就要处理掉一个partition全部的数据，若是内存不够，垃圾回收时是没法回收掉太多对象的，极可能出现OOM异常。因此使用这类操做时要慎重！

6.3 使用foreachPartition替代foreach

原理相似于“使用mapPartitions替代map”，也是一次函数调用处理一个partition的全部数据，而不是一次函数调用处理一条数据。
  在实践中发现，foreachPartitions类的算子，对性能的提高仍是颇有帮助的。好比在foreach函数中，将RDD中全部数据写MySQL，那么若是是普通的foreach算子，就会一条数据一条数据地写，每次函数调用可能就会建立一个数据库链接，此时就势必会频繁地建立和销毁数据库链接，性能是很是低下；  可是若是用foreachPartitions算子一次性处理一个partition的数据，那么对于每一个partition，只要建立一个数据库链接便可，而后执行批量插入操做，此时性能是比较高的。实践中发现，对于1万条左右的数据量写MySQL，性能能够提高30%以上。

6.4 使用filter以后进行coalesce操做

一般对一个RDD执行filter算子过滤掉RDD中较多数据后（好比30%以上的数据），建议使用coalesce算子，手动减小RDD的partition数量，将RDD中的数据压缩到更少的partition中去。
  由于filter以后，RDD的每一个partition中都会有不少数据被过滤掉，此时若是照常进行后续的计算，其实每一个task处理的partition中的数据量并非不少，有一点资源浪费，并且此时处理的task越多，可能速度反而越慢。
  所以用coalesce减小partition数量，将RDD中的数据压缩到更少的partition以后，只要使用更少的task便可处理完全部的partition。在某些场景下，对于性能的提高会有必定的帮助。

6.5 使用repartitionAndSortWithinPartitions替代repartition与sort类操做

repartitionAndSortWithinPartitions是Spark官网推荐的一个算子，官方建议，若是须要在repartition重分区以后，还要进行排序，建议直接使用repartitionAndSortWithinPartitions算子。
  由于该算子能够一边进行重分区的shuffle操做，一边进行排序。shuffle与sort两个操做同时进行，比先shuffle再sort来讲，性能多是要高的。

7. 使用Kryo优化序列化性能

7.1 spark序列化介绍

Spark在进行任务计算的时候，会涉及到数据跨进程的网络传输、数据的持久化，这个时候就须要对数据进行序列化。Spark默认采用Java的序列化器。默认java序列化的优缺点以下:
其好处：
  处理起来方便，不须要咱们手动作其余操做，只是在使用一个对象和变量的时候，须要实现Serializble接口。
其缺点：
  默认的序列化机制的效率不高，序列化的速度比较慢；序列化之后的数据，占用的内存空间相对仍是比较大。

Spark支持使用Kryo序列化机制。Kryo序列化机制，比默认的Java序列化机制，速度要快，序列化后的数据要更小，大概是Java序列化机制的1/10。因此Kryo序列化优化之后，可让网络传输的数据变少；在集群中耗费的内存资源大大减小。

7.2 Kryo序列化启用后生效的地方

Kryo序列化机制，一旦启用之后，会生效的几个地方：
（1）算子函数中使用到的外部变量
  算子中的外部变量可能来着与driver须要涉及到网络传输，就须要用到序列化。
      最终能够优化网络传输的性能，优化集群中内存的占用和消耗
    
（2）持久化RDD时进行序列化，StorageLevel.MEMORY_ONLY_SER
  将rdd持久化时，对应的存储级别里，须要用到序列化。
      最终能够优化内存的占用和消耗；持久化RDD占用的内存越少，task执行的时候，建立的对象，就不至于频繁的占满内存，频繁发生GC。
    
（3）  产生shuffle的地方，也就是宽依赖
  下游的stage中的task，拉取上游stage中的task产生的结果数据，跨网络传输，须要用到序列化。最终能够优化网络传输的性能

7.3 如何开启Kryo序列化机制

// 建立SparkConf对象。
val conf = new SparkConf().setMaster(...).setAppName(...)
// 设置序列化器为KryoSerializer。
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

// 注册要序列化的自定义类型。
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

8. 使用fastutil优化数据格式

8.1 fastutil介绍

fastutil是扩展了Java标准集合框架（Map、List、Set；HashMap、ArrayList、HashSet）的类库，提供了特殊类型的map、set、list和queue；

fastutil可以提供更小的内存占用，更快的存取速度；咱们使用fastutil提供的集合类，来替代本身平时使用的JDK的原生的Map、List、Set.

8.2 fastutil好处

fastutil集合类，能够减少内存的占用，而且在进行集合的遍历、根据索引（或者key）获取元素的值和设置元素的值的时候，提供更快的存取速度

8.3 Spark中应用fastutil的场景和使用

8.3.1 算子函数使用了外部变量

（1）你可使用Broadcast广播变量优化；

（2）可使用Kryo序列化类库，提高序列化性能和效率；

（3）若是外部变量是某种比较大的集合，那么能够考虑使用fastutil改写外部变量；

首先从源头上就减小内存的占用(fastutil)，经过广播变量进一步减小内存占用，再经过Kryo序列化类库进一步减小内存占用。

8.3.2 算子函数里使用了比较大的集合Map/List

在你的算子函数里，也就是task要执行的计算逻辑里面，若是有逻辑中，出现，要建立比较大的Map、List等集合，
可能会占用较大的内存空间，并且可能涉及到消耗性能的遍历、存取等集合操做； 
那么此时，能够考虑将这些集合类型使用fastutil类库重写，

使用了fastutil集合类之后，就能够在必定程度上，减小task建立出来的集合类型的内存占用。 
避免executor内存频繁占满，频繁唤起GC，致使性能降低。

8.3.3 fastutil的使用

第一步：在pom.xml中引用fastutil的包
    <dependency>
      <groupId>fastutil</groupId>
      <artifactId>fastutil</artifactId>
      <version>5.0.9</version>
    </dependency>
    
第二步：平时使用List （Integer）的替换成IntList便可。 
  List<Integer>的list对应的到fastutil就是IntList类型
  
  
使用说明：
基本都是相似于IntList的格式，前缀就是集合的元素类型； 
特殊的就是Map，Int2IntMap，表明了key-value映射的元素类型。

9. 调节数据本地化等待时长

Spark在Driver上对Application的每个stage的task进行分配以前，都会计算出每一个task要计算的是哪一个分片数据，RDD的某个partition；Spark的task分配算法，优先会但愿每一个task正好分配到它要计算的数据所在的节点，这样的话就不用在网络间传输数据；

  可是一般来讲，有时事与愿违，可能task没有机会分配到它的数据所在的节点，为何呢，可能那个节点的计算资源和计算能力都满了；因此这种时候，一般来讲，Spark会等待一段时间，默认状况下是3秒（不是绝对的，还有不少种状况，对不一样的本地化级别，都会去等待），到最后实在是等待不了了，就会选择一个比较差的本地化级别，好比说将task分配到距离要计算的数据所在节点比较近的一个节点，而后进行计算。

9.1 本地化级别

（1）PROCESS_LOCAL：进程本地化
  代码和数据在同一个进程中，也就是在同一个executor中；计算数据的task由executor执行，数据在executor的BlockManager中；性能最好
（2）NODE_LOCAL：节点本地化
  代码和数据在同一个节点中；好比说数据做为一个HDFS block块，就在节点上，而task在节点上某个executor中运行；或者是数据和task在一个节点上的不一样executor中；数据须要在进程间进行传输；性能其次
（3）RACK_LOCAL：机架本地化  
  数据和task在一个机架的两个节点上；数据须要经过网络在节点之间进行传输； 性能比较差
（4）  ANY：无限制
  数据和task可能在集群中的任何地方，并且不在一个机架中；性能最差

9.2 数据本地化等待时长

spark.locality.wait，默认是3s
首先采用最佳的方式，等待3s后降级,仍是不行，继续降级...,最后仍是不行，只可以采用最差的。

9.3 如何调节参数而且测试

修改spark.locality.wait参数，默认是3s，能够增长

下面是每一个数据本地化级别的等待时间，默认都是跟spark.locality.wait时间相同，
默认都是3s(可查看spark官网对应参数说明，以下图所示)
spark.locality.wait.node
spark.locality.wait.process
spark.locality.wait.rack

在代码中设置：
new SparkConf().set("spark.locality.wait","10")

而后把程序提交到spark集群中运行，注意观察日志，spark做业的运行日志，推荐你们在测试的时候，先用client模式，在本地就直接能够看到比较全的日志。 
日志里面会显示，starting task .... PROCESS LOCAL、NODE LOCAL.....
例如：
Starting task 0.0 in stage 1.0 (TID 2, 192.168.200.102, partition 0, NODE_LOCAL, 5254 bytes)

观察大部分task的数据本地化级别 
若是大多都是PROCESS_LOCAL，那就不用调节了。若是是发现，好多的级别都是NODE_LOCAL、ANY，那么最好就去调节一下数据本地化的等待时长。应该是要反复调节，每次调节完之后，再来运行，观察日志 
看看大部分的task的本地化级别有没有提高；看看整个spark做业的运行时间有没有缩短。

注意注意：
在调节参数、运行任务的时候，别本末倒置，本地化级别却是提高了， 可是由于大量的等待时长，spark做业的运行时间反而增长了，那就仍是不要调节了。

10. 基于Spark内存模型调优

10.1 spark中executor内存划分

• Executor的内存主要分为三块

  • 第一块是让task执行咱们本身编写的代码时使用；

  • 第二块是让task经过shuffle过程拉取了上一个stage的task的输出后，进行聚合等操做时使用

  • 第三块是让RDD缓存时使用

10.2 spark的内存模型

在spark1.6版本之前 spark的executor使用的静态内存模型，可是在spark1.6开始，多增长了一个统一内存模型。
  经过spark.memory.useLegacyMode 这个参数去配置
      默认这个值是false，表明用的是新的动态内存模型；
      若是想用之前的静态内存模型，那么就要把这个值改成true。

10.2.1 静态内存模型

实际上就是把咱们的一个executor分红了三部分，
  一部分是Storage内存区域，
  一部分是execution区域，
  还有一部分是其余区域。若是使用的静态内存模型，那么用这几个参数去控制：
  
spark.storage.memoryFraction：默认0.6
spark.shuffle.memoryFraction：默认0.2  
因此第三部分就是0.2

若是咱们cache数据量比较大，或者是咱们的广播变量比较大，
  那咱们就把spark.storage.memoryFraction这个值调大一点。
  可是若是咱们代码里面没有广播变量，也没有cache，shuffle又比较多，那咱们要把spark.shuffle.memoryFraction 这值调大。

• 静态内存模型的缺点

咱们配置好了Storage内存区域和execution区域后，咱们的一个任务假设execution内存不够用了，可是它的Storage内存区域是空闲的，两个之间不能互相借用，不够灵活，因此才出来咱们新的统一内存模型。

10.2.2 统一内存模型

动态内存模型先是预留了300m内存，防止内存溢出。动态内存模型把总体内存分红了两部分，
由这个参数表示spark.memory.fraction 这个指的默认值是0.6 表明另外的一部分是0.4,

而后spark.memory.fraction 这部分又划分红为两个小部分。这两小部分共占总体内存的0.6 .这两部分其实就是：Storage内存和execution内存。由spark.memory.storageFraction 这个参数去调配，由于两个共占0.6。若是spark.memory.storageFraction这个值配的是0.5,那说明这0.6里面 storage占了0.5，也就是executor占了0.3 。

• 统一内存模型有什么特色呢?

Storage内存和execution内存 能够相互借用。不用像静态内存模型那样死板，可是是有规则的

为何受伤的都是storage呢？

是由于execution里面的数据是立刻就要用的，而storage里的数据不必定立刻就要用。

10.2.3 任务提交脚本参考

• 如下是一份spark-submit命令的示例，你们能够参考一下，并根据本身的实际状况进行调节

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 \

10.2.4 我的经验

java.lang.OutOfMemoryError
ExecutorLostFailure
Executor exit code 为143
executor lost
hearbeat time out
shuffle file lost

若是遇到以上问题，颇有可能就是内存除了问题，能够先尝试增长内存。若是仍是解决不了，那么请听下一次数据倾斜调优的课。

吴邪，小三爷，混迹于后台，大数据，人工智能领域的小菜鸟。
更多请关注