1.分配更多资源

1、增加–num-executors、增加–executor-cores，这两个参数增加后能够提升task的并行度
2、增加–executor-memory，在shuffle时reduce操作会先将map端来的数据放在内存上，内存不够用了就往磁盘里存，增大内存减少磁盘IO
3、增加–driver-memory，如果一个RDD数据量过大我们使用了collect操作，就会把数据以数组的形式返回到driver端，导致OOM异常，增大内存避免了该异常

2.提高并行度

1、设置task数量来提高并行度 new SparkConf().set(“spark.defalut.parallelism”,“500”)
2、给RDD重新设置partition的数量 rdd.repartition
3、提高sparksql运行的task数量 设置参数spark.sql.shuffle.partitions=500

3. RDD的重用和持久化

1、可以调用rdd的cache或者persist方法对重复多次使用的RDD进行持久化
2、rdd持久化的时可以采用序列化
（1）如果正常将数据持久化在内存中，那么可能会导致内存的占用过大，这样的话，也许会导致OOM内存溢出。
（2）当纯内存无法支撑公共RDD数据完全存放的时候，就优先考虑使用序列化的方式在纯内存中存储。将RDD的每个partition的数据，序列化成一个字节数组；序列化后，大大减少内存的空间占用。
（3）序列化的方式，唯一的缺点就是，在获取数据的时候，需要反序列化。但是可以减少占用的空间和便于网络传输
（4）如果序列化纯内存方式，还是导致OOM，内存溢出；就只能考虑磁盘的方式，内存+磁盘的普通方式（无序列化）。

4. 广播变量的使用

在很多个task都需要使用同一份数据时，每个task都会复制一份数据，这就导致了不必要的内存开销，如果要持久化一个RDD的话，内存盛不下只能到磁盘，还导致了磁盘IO。

将共用数据使用广播变量广播出去，初始的时候就会在Driver端有一份副本，在某个task要用到该数据的时候就会先去executor的blockmanager中找，blockmanager中没有的话就去Driver端拉取，此后该executor中的task再使用时就只需要从blockmanager中获取。

5. 尽量避免使用shuffle类算子

1、将join改为Broadcast+map不会产生shuffle操作，但仅适用于其中一个RDD数据量比较小的情况
2、使用map端预聚合的算子比如reducebykey和aggregateByKey

6. 使用高性能的算子

1、使用reduceByKey/aggregateByKey替代groupByKey

• reduceByKey/aggregateByKey 可以进行预聚合操作，减少数据的传输量，提升性能
  2、使用mapPartitions替代普通map
• mapPartitions类的算子，一次函数调用会处理一个partition所有的数据，而不是一次函数调用处理一条，缺点是如果一个partition中的数据量过大可能会造成OOM异常
  3、使用foreachPartitions替代foreach
• 原理类似于“使用mapPartitions替代map”，也是一次函数调用处理一个partition的所有数据
  4、使用filter之后进行coalesce操作
• 用于filter算子过滤掉大量数据后，我们需要手动coalesce减少RDD的partition，因为每个partition中的数据不是很多，有点资源浪费，这时即使有多个task并行执行也许也没有多快反而会变慢
  5、使用repartitionAndSortWithinPartitions替代repartition与sort类操作
• 用于重分区后还需要进行排序的情况

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

Kryo序列化机制，比默认的Java序列化机制，速度要快，序列化后的数据要更小，大概是Java序列化机制的1/10。
keyo生效的地方：

1. 算子函数中使用到的外部变量
2. 持久化RDD时进行序列化，StorageLevel.MEMORY_ONLY_SER
3. 产生shuffle的地方，也就是宽依赖

// 设置序列化器为KryoSerializer。
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
// 注册要序列化的自定义类型。
conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

8. 使用fastutil优化数据格式

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

fastutil能够提供更小的内存占用，更快的存取速度；我们使用fastutil提供的集合类，来替代自己平时使用的JDK的原生的Map、List、Set.
2、fastutil好处
fastutil集合类，可以减小内存的占用，并且在进行集合的遍历、根据索引（或者key）获取元素的值和设置元素的值的时候，提供更快的存取速度；
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分配到距离要计算的数据所在节点比较近的一个节点，然后进行计算。

1. 本地化级别

• PROCESS_LOCAL：进程本地化，在同一个executor中
• NODE_LOCAL：节点本地化
• RACK_LOCAL：机架本地化
• ANY：无限制

2. 数据本地化等待时长
   spark.locality.wait，默认是3s
3. 如何调节参数并且测试
   
   在代码中设置：
   new SparkConf().set(“spark.locality.wait”,“10”)

然后把程序提交到spark集群中运行，注意观察日志，spark作业的运行日志，推荐大家在测试的时候，先用client模式，在本地就直接可以看到比较全的日志。
看到日志中大多数的PROCESS_LOCAL就不用了调节了，如果大多是NODE_LOCAL、ANY就需要调一下，但需要注意的是不要因为调节等待时长反而导致任务变慢了

10. 基于Spark内存模型调优

内存模型分为两种，在spark1.6版本以前，spark的executor使用的静态内存魔模型，在spark1.6开始，多增加了一个统一内存模型。可以通过spark.memory.useLegacyMode这个参数去配置。

• 静态内存模型
  
  如果我们cache数据量比较大， 或者是我们的广播变量比较大，那我们就把spark.storage.memoryFraction这个值调大一点
  但如果我们代码里没有广播变量，也没有cache，shuffle又比较多，那我们要把spark.shuffle.memoryFraction这个值调大

• 静态内存模型的缺点
  我们配置好内存占比后，假设一个任务execution内存不够用了，但Storage内存区域是空闲的，两者之间不能相互借用，不够灵活

• 统一内存模型
  
  动态内存模型先是预留了300M内存，防止内存溢出，动态内存模型把整体内存分为统一内存、其他内存两部分，由spark.memory.fraction控制，默认是0.6
  然后spark.memory.fraction 这部分又划分成为两个小部分。这两小部分共占整体内存的0.6。其实就是Storage内存和execution内存，由spark.memory.storageFraction这个参数控制。

特点：Storage内存和execution内存 可以相互借用。不用像静态内存模型那样死板，但是是有规则的，就是以execution为主，execution使用的内存多了就会把Storage的内存数据写到磁盘上。

./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 \