在SPARK中,主要有三个地方涉及到了序列化:
- 在算子函数中使用到外部变量时,该变量会被序列化后进行网络传输(见“原则七:广播大变量”中的讲解)。
- 将自定义的类型作为RDD的泛型类型时(比如JAVARDD,Student是自定义类型),所有自定义类型对象,都会进行序列化。因此在这种情况下,也要求自定义的类必须实现Serializable接口。
- 使用可序列化的持久化策略时(比如MEMORY_ONLY_SER),Spark会将RDD中的每个partition都序列化成一个大的字节数组。
对于这三种出现序列化的地方,我们都可以通过使用Kryo序列化类库,来优化序列化和反序列化的性能。Spark默认使用的是Java的序列化机制,也就是ObjectOutputStream/ObjectInputStream API来进行序列化和反序列化。但是Spark同时支持使用Kryo序列化库,Kryo序列化类库的性能比Java序列化类库的性能要高很多。官方介绍,Kryo序列化机制比Java序列化机制性能要高十倍左右。Spark之所以默认没有使用Kryo作为序列化类库,是因为Kryo要求最好要注册所有需要进行序列化的自定义类型,因此对于开发者来说,这种方式比较麻烦。
以下是使用Kryo的代码示例,我们只要设置序列化类,再注册要序列化的自定义类型即可(比如算子函数中使用到的外部变量类型、作为RDD泛型类型的自定义类型等);
//创建SparkConf对象
val conf = new SparkConf().setMaster(...).setAppName(...)
//设置序列化器为KryoSerializer
conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer)
//注册要序列化的自定义类型
conf.registerKryoClasses(Array(classOf[MyClass1] , classOf[MyClass2]))