性能优化概览
why
Spark是基于内存的计算,所以集群的CPU、网络带宽、内存等都可能成为性能的瓶颈。
when
Spark应用开发成熟时,满足业务要求后,就可以开展性能优化了。
what
一般来说,Spark应用程序80%的优化集中在内存、磁盘IO、网络IO,即Driver、Executor的内存、shuffle的设置、文件系统的配置,集群的搭建,集群和文件系统的搭建(文件系统的集群在同一个局域网内)。
how
web UI+log是Spark性能优化的倚天剑和屠龙刀。
driver的log信息大致如“INFO BlockManagerMasterActor: Added rdd_0_1 in memory on mbk.local:50311 (size: 717.5 KB, free: 332.3 MB)”的日志信息。这就显示了每个partition占用了多少内存。
内存都去哪了
Java对象头
每个Java对象,都有一个对象头,会占用16个字节,主要是包括了一些对象的元信息,比如指向它的类的指针。如果一个对象本身很小,比如就包括了一个int类型的field,那么它的对象头实际上比对象自己还要大。
String对象
Java的String对象会比它内部的原始数据多出40个字节。因为它内部使用char数组来保存内部的字符序列的,并且还得保存诸如数组长度之类的信息;而且String使用的是UTF-16编码,每个字符会占用2个字节。比如,包含10个字符的String,会占用60个字节。
集合类型
Java中的集合类型,比如HashMap和LinkedList,内部使用的是链表数据结构,所以对链表中的每一个数据,都使用了Entry对象来包装。Entry对象不光有对象头,还有指向下一个Entry的指针,通常占用8个字节。
其他
元素类型为原始数据类型(比如int)的集合,内部通常会使用原始数据类型的包装类型,比如Integer,来存储元素。
List<Integer> list = new ArrayList<Integer>()
性能优化方法
数据序列化
Spark默认序列化机制
Spark自身对于序列化的便捷性和性能进行了一个取舍和权衡。默认,Spark倾向于序列化的便捷性,使用了Java自身提供的序列化机制——基于ObjectInputStream和ObjectOutputStream的序列化机制。
Java序列化机制的缺陷
Java序列化机制的性能并不高,序列化的速度相对较慢;而且序列化以后的数据,还是相对来说比较大,还是比较占用内存空间。
Kryo序列化机制
Spark也支持使用Kryo类库来进行序列化。Kryo序列化机制比Java序列化机制更快,而且序列化后的数据占用的空间更小,通常比Java序列化的数据占用的空间要小10倍。SparkConf().set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
Kryo使用场景
算子函数使用到了外部的大数据的情况。
比如自定义了一个MyConfiguration对象,里面包含了100m的数据。然后,在算子函数里面,使用到了这个外部的大对象。
conf.registerKryoClasses(XXX.class)
优化Kryo缓存大小
如果注册的要序列化的自定义的类型,本身特别大,就需要调整Kryo缓存的大小,默认值是2M。SparkConf.set(“spark.kryoserializer.buffer.mb”,nM)。
数据结构优化
场景
算子中用到的内部和外部的数据,优化之后,会减少内存的消耗和占用。
优先使用数组以及字符串而不是集合类
比如将
List<Integer> list = new ArrayList<Integer>()
替换为
int[] arr = new int[]
这样array既比List少了额外信息的存储开销,还能使用原始数据类型(int)来存储数据,要节省内存的多。
将
Map<Integer, Person> persons = new HashMap<Integer, Person>()
优化为特殊的字符串格式
id:name,address|id:name,address...。
避免使用多层嵌套的对象结构
如
public class Teacher {
private List<Student> students = new ArrayList<Student>()
}
就是非常不好的例子。因为Teacher类的内部又嵌套了大量的小Student对象。优化为json字符串来存储数据
{
"teacherId": 1,
"teacherName": "leo",
students:[
{"studentId": 1, "studentName":"tom"},
{"studentId":2, "studentName":"marry"}
]
}
尽量使用int替代String
如用int行ID替代UUID等。
RDD持久化
持久化的场景
对RDD反复使用和重要的、关键的、耗时长的RDD。
持久化方法
使用cache()|persist()方法进行持久化,使用unpersist()方法取消持久化。
持久化策略
Spark提供的多种持久化级别,主要是为了在CPU和内存消耗之间进行取舍。优先使用MEMORY_ONLY,内存不足时使用MEMORY_ONLY_SER。
注意事项
JavaRDD<String> targetwords = words.filter(new Function<String, Boolean>() {}).cache();
不应该是
JavaRDD<String> targetwords = words.filter(new Function<String, Boolean>() {});
targetwords.cache();
Spark自己也会在shuffle操作时进行数的持久化,主要是为了在节点失败时避免重算整个过程。
提高并行度
Spark集群的资源并不一定会被充分利用到,所以要尽量设置合理的并行度,来充分地利用集群的资源,以充分提高Spark应用程序的性能。
Spark会自动设置以文件作为输入源的RDD的并行度,依据其大小,比如HDFS,就会给每一个block创建一个partition,也依据这个设置并行度。对于reduceByKey等会发生shuffle的操作,就使用并行度最大的父RDD的并行度即可。
手动使用textFile()、parallelize()等方法的第二个参数来设置并行度;
使用spark.default.parallelism参数来设置统一的并行度
Spark官方的推荐是,给集群中的每个cpu core设置2~3个task。
比如说,spark-submit设置了executor数量是10个,每个executor要求分配2个core,那么application总共会有20个core。此时可以设置new SparkConf().set("spark.default.parallelism", "60")
来设置合理的并行度,从而充分利用资源。
广播共享数据
优化前
默认情况下,算子函数使用到的外部数据,会被拷贝到每个task中,如果使用到的外部数据很大,那么就会占用大量的内存空间和网络传输。
优化后
外部数据在每个节点上只保留一份副本,大大节省了内存和网络传输。
广播共享数据的用户
创建广播变量
...
Broadcast<T> broadcast = sc.broadcast(T);
...
使用广播变量
...
broadcast.value();
...
数据本地化
数据本地化对性能的影响
数据本地化对于Spark Job性能有着巨大的影响,如果数据与要计算它的代码是在一起的,那么性能当然会非常高。Spark倾向于使用最好的本地化级别来调度task,如果没有任何未处理的数据在空闲的executor上,那么Spark就会放低本地化级别。这时有两个选择:等待直到executor上的cpu释放出来,那么就分配task过去或者立即在任意一个executor上启动一个task。
数据本地化级别
PROCESS_LOCAL:数据和计算它的代码在同一个JVM进程中。
NODE_LOCAL:数据和计算它的代码在一个节点上,但是不在一个进程中;
NO_PREF:数据从哪里过来,性能都是一样的。
RACK_LOCAL:数据和计算它的代码在一个机架上。
ANY:数据可能在任意地方,比如其他网络环境内,或者其他机架上。
优化参数
spark.locality.wait(3000毫秒)
spark.locality.wait.node
spark.locality.wait.process
spark.locality.wait.rack
reduceByKey和groupByKey优化
如果能用reduceByKey,那就用reduceByKey,因为它会在map端,先进行本地combine,可以大大减少要传输到reduce端的数据量,减小网络传输的开销。
只有在reduceByKey处理不了时,才用groupByKey().map()来替代。
JVN垃圾回收调优
GC对性能的影响
默认情况下,Executor的内存空间60%用于RDD的缓存,40%分配给Task用于运行。Task很可能很快就耗光了内存而触发GC。GC发生时将停止一切工作线程,GC本身需要花费时间,如果再频繁发生GC,将严重影响Spark应用程序的性能。
GC 优化
可通过调整比例达到优化GC的目的。
SparkConf().set(“spark.storage.memoryFraction”, “0.5”)
比值在0.6~0.1之间调整。
若配合使用序列化持久化级别如MEMORY_ONLY_SER何kryo等手段,将会有更好的性能优化。
shuffle优化
spark.shuffle.consolidateFiles:是否开启shuffle block file的合并,默认为false
spark.reducer.maxSizeInFlight:reduce task的拉取缓存,默认48m
spark.shuffle.file.buffer:map task的写磁盘缓存,默认32k
spark.shuffle.io.maxRetries:拉取失败的最大重试次数,默认3次
spark.shuffle.io.retryWait:拉取失败的重试间隔,默认5s
spark.shuffle.memoryFraction:用于reduce端聚合的内存比例,默认0.2,超过比例就会溢出到磁盘上