Spark 调优

大家都知道spark的计算是基于内存的，所以内存的合理使用对spark调优至关重要，其实大方向主要关注三个方面CPU，网络带宽，内存。下文主要覆盖两个方面，一个是网络（数据序列化），一个是内存。

数据序列化

序列化在分布式系统中扮演着和重要的角色，通常是优化的第一件事。spark为了性能和便捷性（允许在程序中使用任意的java对象），提供了两种序列化的方式：

• java serialization 默认的序列化方式，使用了java的ObjectOutputStream框架，代码中仅仅需要implement java.io.Serializable就可以了，虽然提供了灵活性但是序列化的结果太大。
• Kryo serialization spark提供了Kryo的库序列化的更快。kryo的序列化比java序列化更加快速和简洁，但是并不是支持所有的序列化的类型，需要你注册你所使用的类。

设置SparkConf通过conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer").序列化不仅对不同node间传输数据有效，还可以序列化rdd到磁盘。从spark2.0开始，当rdd中包含简单的类型和他们的数组，字符串的时候，默认使用kyro序列化。
可以利用kyro注册自定义的类，conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))
如果你的对象非常大，你需要增加spark.kryoserializer.buffer.max=64m,保证有足够的空间放入序列化的对象。如果没有注册自定义的类，kryo将会保存每个对象的整个类名，空间显然被浪费了。

内存调优

在内存调优方面需要考虑三个方面：对象使用的内存空间，使用这些对象使用的内存空间，GC使用的空间。
默认情况下，java对象是很容易被使用的，但是也很容易使用原始数据大小2~5倍的内存空间，大致是因为以下几个原因：

• 每个java对象都有一个对象头，包含16个字节和一些必要的信息（指针），对于一些小的对象来说，对象头比对象本身都大。
• java String对象大约比原始string多40字节。对于一个10个字符的字符串，很容易占据50~60bytes的空间。
• 对于一般集合类，例如HashMap和LinkedList使用了链表，每个对象不仅有对象头，还有指向下一个的指针，占用了大量的空间。

内存管理概况

spark内存使用却大多数被消耗在两个方面：执行和存储。执行内存是指shuffle时的计算，join，sort，aggregation。存储内存是指cache和集群中数据的传输。执行和存储共享同一个内存（内存统一管理），但是当执行内存正在执行复杂的运算，存储内存就不可以占用执行内存。*统一内存管理的好处是不需要用户去担心怎么划分两者。
虽然spark提供了两个参数去修改两者，但是默认的值可以覆盖大多数的情况。

• spark.memory.fraction描述了统一内存占用JVM堆内存的比例（60%），其余的用来运行spark内部的数据结构和元信息。此值应该被设置，配合GC的老年代和永久代使用
• spark.memory.storageFraction是不可被执行内存共享的存储内存占用统一内存的比例。

内存的使用

计算一个数据集大小的最好方式就是将数据集保存为一个rdd，将其放入cache中，从webUI观察Storage的大小 或者可以使用SizeEstimator类中的estimate方法，可以计算广播变量所占用executor的内存。

数据结构调优

首先应该避免java对象过多的携带其他与数据无关的信息。

1. 保证自己的数据结构对象数据的大小被提前分配，尽量使用原始类型，避免使用java和scala的集合类。fastutil提供了与java标准库兼容的原始类型。
2. 避免多层小对象的嵌套
3. 对于key来说，尽量使用数字或者枚举类型。
4. 如果内存少于32G，设置JVM的flag -xx:UseCompressedOops确保使用4个字节的指针而不是八个字节。 export SPARK_JAVA_OPTS="-xx:UseCompressedOops"

序列化rdd的存储

如果对数据结构优化之后，对象仍然很大，就需要以序列化的形式保存Rdd到内存以减少内存的使用。可以通过设置StorageLevel=MEMORY_NOLY_SER，将每个rdd partition作为一个大的字节数组。不过这种方式有一个缺点，就是不同node使用的时候需要先反序列化数组，然后使用它。强烈建议使用Kyro序列化

GC优化

在程序中有许多暂时使用的对象的时候，JVM的GC可能是个问题。需要指出的是GC的代价和java对象的大小是正比，解决GC最好的方式就是使用序列化。使用序列化来cache数据。
造成GC的原因可能是执行内存和存储内存的干扰，下面就讨论如何减小这种影响：

检测GC的影响

首先需要知道GC的频率和GC所占用的时间。通过在JVM中添加-verbose：gc -XX:+PrintGCDetail -XX:+PrintGCTimeStamps.这样GC的信息就被打印到了每个work node上，而不是driver上。

高级GC调优

在调优之前需要了解一下JVM内存管理基本的原理

• Java堆被分配为两个部分年轻代和老年代。年轻代保存短存活的对象，老年代保存较长的生命周期的对象。
• 年轻代被分成三个部分[Eden，S1，S2]
• 简单描述一下GC的过程：当Eden满了，一个minor GC就将活的对象从Eden和S1移到了S2中。Survivor区域进行置换，如果一个对象足够老或者S2已经满了，对象将被移到老年代、如果老年代满了，full GC将会被调用。
  GC调优的目标是确保长存活的RDD被保存在老年代中，年轻代有足够的空间保存短存活的对象。这样就可以避免full GC临时的对象。
• 如果观察到一个task的完成伴随着对此full GC，那就意味着执行此task的executor没有足够的内存。
• 如果是频繁的minor GC但是没有full GC，分配足够的内存到Eden，设置年轻代的内存大小为-Xmn = 4/3*E，E为预估Eden的大小
• 观察打印的GC的信息，如果老年代即将被full GC,通过spark.memory.fraction减少caching的内存大小。
• 使用-XX:UseG1GC,提高GC的效率。
• 如果数据是从hdfs上读取，可以估算数据占用的内存大小。
  根据经验，GC调优取决于你的程序和可用内存数量。GC对每个executor调优可以设置spark.executor.extraJavaOptions.

其他方面

• 并行度 Spark自动计算map task的数量通过数据的大小。对于reduce task使用父RDD中大的partition number的那一个。如果rdd没有父rdd的话就需要通过spark.default.parallelism来设置。建议每个core跑2~3个tasks。
• reduce task的内存使用 有时会有OOM的错误，并不是因为你的内存不足以放下你的数据。而是对于reduce task将会在每个task中建立一个hash table（很大）。最简单的处理就是增加task的并行度。spark可以在
  一个executor中高效的完成200ms以内的task，多个tasks可以重用同一个executor。因此可以放行的增加并行度。
• 广播变量 使用broadcast可以极大地减少序列化task的大小和传输一个task到executor的代价。如果你的程序使用了一个大的对象在driver端。就需要是用broadcast来优化
• 数据本地性 数据本地性对spark性能的影响很大。如果数据和代码在一起就会执行的很快，如果数据和代码不在一起，一个必须被传输。一般是代码被传输。spark schedule一般基于数据本地性。数据本地性就是指优先处理近的数据。

PROCESS_LOCAL
NODE_LOCAL
NO_PREF
RACK_LOCAL
ANY

Spark优先使用数据本地性，但是本不总是这样。如果executor没有未被处理的数据在这个executor上，并且超时，就会使用低一级的方式。spark先等CUP空闲，一旦超时，就会将数据移动到另一个executor中。这个时间可以被控制通过spark.locality参数。

总结

一般情况下，设置Kyro序列化和以序列化的形式persist数据，将会解决绝大多数的问题。