广播变量
在Spark Application中,经常会使用到一个共享变量,众所周知的,Spark是一个并行计算框架,对于这个变量,每一个executor的task在访问它的时候,都会去拷贝一份副本去使用。如下图所示:
对于这种默认方式,它会极大的系统的内存,我们可以假设一个集群中有1024个task,这个共享变量大小假设为1M,那么就会去复制1024份到集群上去,这样就会有1个G的数据在网络中传输,并且系统需要耗费1G内存去为这些副本分配空间,这样对于系统有什么影响呢?
如果系统内存不足,RDD持久化的时候无法在内存中持久化,需要持久化到磁盘中,那么后续的操作会因为频繁的磁盘IO使得速度变慢,性能下降。
当task中创建对象时,发现堆中内存不足,那么就需要进行GC操作,进行GC的时候,会导致工作线程暂停,如果内存严重不足,频繁的GC对于Spark作业的速度的影响是可想而知的。
对于以上问题,我们该怎么进行性能优化呢?
这时,我们可以使用Broadcast,将这种每个task需要用到的共享变量广播出去。
从上面的图中可以看到,当每一个task需要使用这个变量的时候都会拷贝一份。如果使用广播变量,首先该广播变量会拷贝一份副本到Driver中,当每一个executor的task使用到该变量时,首先会去每个executor的BlockManager中去检查是否有该变量的副本,如果没有,接着会去Driver中去拷贝一份副本到BlockManager中,然后供该executor中的每一个task使用,到下一个executor的task需要使用这个变量时,它的BlockManager可以去Driver中拷贝副本,也可以去距离比较近的executor的BlockManager中去拷贝。(每一个executor中的BlockManager的作用是负责管理每一个executor对应的内存和磁盘的数据。)其原理图如上所示:
在默认情况下,如果是1024个task需要消耗1G内存,但是如果我们有50个executor来平分这些task,那么只需要50个副本即可,总共消耗了50M内存,那么在内存的消耗了节省了大约20倍。而且副本的复制有时不需要从Driver拷贝,而是从其他executor中拷贝,那么,网络 传输带来的性能消耗也会小很多,可想而知,使用广播变量可以节省很多内存,从而使得性能显著提升。
如何使用广播变量呢?
比如我们的共享变量是一个map类型的变量,我们可以使用Spark上下文来创建广播变量:
Broadcast<Map<String>> broadcast=sc.broadcast(map);
在task中使用的时候可以使用value方法或者getValue方法来获取它的值:
Map<String> map=broadcast.value;
Kryo序列化
在上面通过广播变量降低网络传输压力以及节省了不少内存之后,我们可以再进一步的优化,序列化是一种不错的选择,可以减少数据占用的内存大小。
默认的Spark使用的是java序列化机制,即通过ObjectOutputStream / ObjectInputStream,对象输入输出流机制,来进行序列化。
使用默认的这种序列化机制,好处在于简单方便,不需要你自己进行任何配置,只需要在需要序列化的类上实现Serializable接口;缺点在于序列化的效率不高,序列化的速度比较低,序列化之后的数据占用空间依旧很大。因此,我们在必要时可以手动指定序列化方式进行优化。
Spark支持Kryo序列化机制,Kryo序列化机制,比默认的Java序列化机制,速度要快,序列化后的数据要更小,大概是Java序列化机制的1/10。
因此,使用Kryo序列化机制可以让 网络中传输的数据更小,而且在集群中耗费的内存也大大减少。
Kryo序列化作用的地方
Kryo序列化一旦启用,在以下几个地方将会生效:
算子函数中使用的外部变量。算子函数中使用到的外部变量,使用Kryo以后,优化网络传输的性能,可以优化集群中内存的占用和消耗
持久化RDD时进行序列化,比如StorageLevel.MEMORY_ONLY_SER。持久化RDD,优化内存的占用和消耗;持久化RDD占用的内存越少,task执行的时候,创建的对象,就不至于频繁的占满内存,频繁发生GC。
shuffle时进行序列化,可以优化网络传输的性能。
如何使用Kryo序列化
第一步:在SparkConf中设置序列化属性spark.serializer,值为org.apache.spark.serializer.KryoSerializer。
第二步:注册你需要使用Kryo序列化的一些自定义类,使用SparkConf.registerKryoClasses()
方法进行注册。
例如:
new SparkConf()
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(new Class[]{CategorySortKey.class})
使用广播变量和序列化很简单,但是我们不能忽视它对于系统性能的影响,也许一个小小的修改就能对系统的性能提升很多倍,这也是我们性能优化的重点,不能忽略每一个细节,一昧的去追去高大上的优化技巧。