Spark 调优

Spark 调优

问题：只要会用就可以，为什么还要精通内核源码与调优？
Spark 性能优化概览：
Spark的计算本质是，分布式计算
所以，Spark程序的性能可能因为集群中的任何因素出现瓶颈：CPU、网络带宽、或者内存

CPU、网络带宽，是运维来维护的
聚焦点：内存

如果内存能够容纳下所有的数据，那就不需要调优了
如果内存比较紧张，不足以放下所有数据（10亿量级—500G）,需要对内存的使用进行性能优化
比如：使用某些方法减少内存的消耗

Spark性能优化，主要针对在内存的使用调优

Spark性能优化的技术：
1、使用高性能序列化类库
2、优化数据结构
3、对于多次使用的RDD进行持久化、checkpoint
4、持久化级别：MEMORY_ONLY —> MEMORY_ONLY_SER 序列化
5、Java虚拟机垃圾回收调优
6、Shuffle调优，1.x版本中，90%的性能问题，都是由于Shuffle导致的。

其他性能优化：
1、提高并行度
2、广播共享数据
等等。。。

一、诊断Spark内存使用

首先要看到内存使用情况，才能进行针对性的优化

1、内存花费

（1）每个Java对象，都有一个对象头，占用16字节，包含一些对象的元信息，比如指向他的类的指针
如果对象本身很小，比如int，但是他的对象头比对象自己还大

（2）Java的String对象，会比他内存的原始数据，多出40个字节
String内部使用的char数组来保存内部的字符串序列，并且还要保存诸如输出长度之类的信息
char使用的是UTF-16编码，每个字符会占2个字节。比如，包含10个字符的String，2*10+40=60字节

（3）Java中的集合类型，比如HashMap和LinkedList，内部使用链表数据结构
链表中的每个数据，使用Entry对象包装
Entry对象，不光有对象头，还有指向下一个Entry的指针，占用8字节

（4）元素类型为原始数据类型（int），内部通常会使用原始数据类型的包装类型（Integer）来存储元素

2、如何判断Spark程序消耗内存情况？

预估
（1）设置RDD的并行度
两种方法创建RDD，parallelize() textFile() 在这两个方法中，传入第二个参数，设置RDD的partition数量
在SparkConfig中设置一个参数：
spark.default.parallelism
可以统一设置这个application中所有RDD的partition数量

（2）将RDD缓存 cache()

（3）观察日志：
driver的日志
/root/hd/spark-2.1.0-bin-hadoop2.7/work

Master和Worker的日志
/root/hd/spark-2.1.0-bin-hadoop2.7/logs

scala> val rdd1 = sc.textFile("/root/hd/tmp_files/test_Cache.txt")
rdd1: org.apache.spark.rdd.RDD[String] = /root/hd/tmp_files/test_Cache.txt MapPartitionsRDD[1] at textFile at <console>:24

scala> rdd1.cache
res1: rdd1.type = /root/hd/tmp_files/test_Cache.txt MapPartitionsRDD[1] at textFile at <console>:24

scala> rdd1.count
res2: Long = 921911                                                             

scala> rdd1.count
res3: Long = 921911

/root/hd/spark-2.1.0-bin-hadoop2.7/work
19/04/17 17:02:10 INFO MemoryStore: Block rdd_1_1 stored as values in memory (estimated size 22.9 MB, free 343.1 MB)
19/04/17 17:02:10 INFO MemoryStore: Block rdd_1_0 stored as values in memory (estimated size 22.9 MB, free 320.2 MB)

（4）将这个内存信息相加，就是RDD内存占用量
22.9MB+22.9MB=45.8MB

二、使用高性能序列化类库

1、数据序列化概述

数据序列化，就是将对象或者数据结构，转换成特定的格式，使其可在网络中传输，或存储在内存或文件中
反序列化，是相反的操作，将对象从序列化数据中还原出来
序列化后的数据格式，可以是二进制，xml，Json等任何格式
对象、数据序列化的重点在于数据的交换与传输

在任何分布式系统中，序列化都是扮演着一个重要的角色
如果使用的序列化技术，操作很慢，或者序列化后的数据量还是很大，会让分布式系统应用程序性能下降很多
所以，Spark性能优化的第一步，就是进行序列化的性能优化

Spark自身默认会在一些地方对数据进行序列化，比如Shuffle。另外，我们使用了外部数据（自定义类型），也要让其课序列化

Spark本身对序列化的便捷性和性能进行了取舍
默认情况下：Spark倾向于序列化的便捷性，使用了Java自身提供的序列化机制，很方便使用

但是，Java序列化机制性能不高，序列化速度慢，序列化后数据较大，比较占用内存空间

2、kryo

Spark支持使用kryo类库来进行序列化
速度快，占用空间更小，比Java序列化数据占用空间小10倍

3、如何使用kryo序列化机制

（1）设置Spark conf
bin/spark-submit will also read configuration options from conf/spark-defaults.conf,
in which each line consists of a key and a value separated by whitespace. For example:

spark.master spark://5.6.7.8:7077
spark.executor.memory 4g
spark.eventLog.enabled true
spark.serializer org.apache.spark.serializer.KryoSerializer

（2）使用kryo是，要求需要序列化的类，要提前注册，以获得高性能
conf.registerKryoClasses(Array(classOf[Count],……))
conf.registerKryoClasses(Array(classOf[类], classOf[类], ……))

4、kryo类库的优化

（1）优化缓存大小
如果注册的自定义类型，本身特别大（100个字段），会导致要序列化的对象太大
此时需要对kyro本身进行优化。因为kryo内部的缓存，可能不能存放这么大的class对象
spark.kryoserializer.buffer.max 设置这个参数，将其调大

（2）预先注册自定义类型
虽然不注册自定义类型，kryo也可以正常工作，但会保存一份他的全限定类名，耗费内存
推荐预先注册要序列化的自定义类型

三、优化数据结构

1、概述

要减少内存的消耗，除了使用高效的序列化类库外，还要优化数据结构
避免Java语法特性中所导致的额外内存开销

核心：优化算子函数内部使用到的局部数据或算子函数外部的数据
目的：减少对内存的消耗和占用

2、如何做

（1）优先使用数组以及字符串，而不是集合类。即：优先使用Array，而不是ArrayList、LinkedList、HashMap
使用int[] 会比List<Integer> 节省内存

（2）将对象转换成字符串
企业中，将HashMap、List这种数据，统一用String拼接成特殊格式的字符串

Map<Integer,Person> persons = new HashMap<Integer,Person>()

可以优化为：
“id:name,address”
String persons = “1:Andy,Beijing|2:Tom,Tianjin….”

（3）避免使用多层嵌套对象结构
举例：
下面的例子不好，因为Teacher类的内部又嵌套了大量的小的Student对象
public class Teacher{ private …..; privage List<Student> students = new ArrayList()}

解决：转换成字符串进行处理
{“teacherId”: 1, “students”:[{“stuId”:1…..},{}]}

（4）对于能够避免的场景，尽量使用int代替String
虽然String比List效率高，但int类型占用更少内存

比如：数据库主键，id，推荐使用自增的id，而不是uuid

四、rdd.cache checkpoint

五、持久化级别

MEMORY_ONLY —> MEMORY_ONLY_SER 序列化

六、Java虚拟机的调优

1、概述

如果在持久化RDD的时候，持久化了大量的数据，那么Java虚拟机的垃圾回收就可能成为一个瓶颈

Java虚拟机会定期进行垃圾回收，此时会追踪所有Java对象，并且在垃圾回收时，找到那些已经不再使用的对象

清理旧对象，给新对象腾出空间

垃圾回收的性能开销，是与内存中的对象数量成正比

在做Java虚拟机调优之前，必须先做好上面的调优工作，这样才有意义

必须注意顺序（先进行完上面的调优，再进行JVM调优）

2、Spark GC原理

垃圾回收期GC
垃圾回收器，寻找那些对象已经不再使用，将其清除出去

GC对性能的影响在于，如果内存中数据量较大，会很频繁地造成内存空间不够，导致GC频繁发生，而GC本身是有性能消耗的，如果频繁发生，对性能影响严重

此外，如果数据量过大，每次要回收的数据量也很大，导致GC慢

另外GC发生的时候，GC是一个线程，我们Task运行时线程叫做工作线程。GC运行时会让工作线程停下来，让GC单独运行，影响Spark应用程序的运行速度，降低了性能

核心：不让GC频繁发生

3、监测垃圾回收

我们可以进行监测，比如多久进行一次垃圾回收以及耗费的时间等等。

spark-submit脚本中，添加一个配置
–conf “spark.executor.extraJavaOptions=-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimesStamps”

注意：这个是输出到worker日志中，而不是driver日志
/root/hd/spark-2.1.0-bin-hadoop2.7/logs worker日志
/root/hd/spark-2.1.0-bin-hadoop2.7/work driver日志

4、优化Executor内存比例

目的：减少GC次数

对于GC调优来说，最重要的就是调节，RDD的缓存占用的内存空间与算子执行时创建对象所占用的内存空间的比例

Executor:Task=3:2 （Executor 占60%给RDD缓存，Task占40%）
对于默认情况，Spark使用每个Executor 60% 的内存空间来缓存RDD，在task运行期间所创建的对象，只有40%内存空间来存放

在默认情况下，很可能发生的事情，分配给Task的内存不够，
导致新创建对象时，很快占满内存，GC启动，找到不再使用的对象，清楚内存

所以，如果Task分配内存过小，可能会导致GC频繁发生，工作线程停止

可以通过调节比例，将RDD缓存空间占比调节到40%，降低Task GC频率

需要配合其他优化：
kryo优化
持久化级别优化
数据结构优化

使用：conf.set(“spark.storage.memoryFraction”,0.5)

5、Java GC 调优 （-）

让GC更快的处理完

七、shuffle原理

1、优化前

假设一个节点上，有4个shufflemaptask，有两个CPU core

每个MapTask会为每个ReduceTask创建一份bucket缓存，以及对应的ShuffleBlockFile磁盘文件

问题：假设有100个MapTask，100个ReduceTask，会产生100*100即10000个文件，磁盘io过多，影响性能

假设另外一个节点上，运行了4个ReduceTask

每个ReduceTask拉取过来的数据，其实会组成内部的RDD，叫ShuffledRDD，优先放入内存，如果不够，写入磁盘

每个ReduceTask针对数据进行聚合，最后生成MapPartitionsRDD，执行reduceByKey操作希望的到那个RDD

2、优化后

在Spark新版本中，引入了consolidation的机制，提出了ShuffleGroup概念

一个ShuffleMapTask执行完后，写入本地文件不会变，但那时，下一个ShuffleSMapTask运行的时候，可以直接将数据写入之前的本地文件

相当于，多个ShuffleMapTask的输出进行了合并，大大减少文件数量

1和2卡哇伊乘坐一组ShuffleGroup，每个文件中，都存储了多个ShuffleMapTask的数量，每个ShuffleMapTask的数据叫做segment，通过外部索引，来标记每个ShuffleMapTask的数据以及偏移量，对不同的ShuffleMapTask的数据进行区分

八、其他调优

1、提高并行度

2、广播共享数据