大家都知道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对象过多的携带其他与数据无关的信息。
- 保证自己的数据结构对象数据的大小被提前分配,尽量使用原始类型,避免使用java和scala的集合类。fastutil提供了与java标准库兼容的原始类型。
- 避免多层小对象的嵌套
- 对于key来说,尽量使用数字或者枚举类型。
- 如果内存少于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数据,将会解决绝大多数的问题。