Spark的Kryo序列化注册

喜欢ヅ旅行 2022-07-13 14:41 203阅读 0赞

Spark的Kryo序列化注册  
Spark序列化可以将RDD序列化来减少内存占用。 对于优化网络性能极为重要  
1、java序列化：

spark.serializer=org.apache.spark.serializer.JavaSerialization

Spark默认 使用Java自带的ObjectOutputStream 框架来序列化对象，这样任何实现了 java.io.Serializable 接口的对象，都能被序列化。  
Java序列化很灵活但性能差速度很慢，同时序列化后占用的字节数也较多。  
2、Kryo序列化：

spark.serializer=org.apache.spark.serializer.KryoSerialization

KryoSerialization速度快,产生的结果更为紧凑（通常能提高10倍）。  
但Kryo不支持所有实现了java.io.Serializable 接口的类型，它需要你在程序中 register 需要序列化的类型，以得到最佳性能。  
  
  
在 SparkConf 初始化的时候调用

conf.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”)

使用 Kryo。这个设置不仅控制各个worker节点之间的混洗数据序列化格式，同时还控制RDD存到磁盘上的序列化格式。  
需要在使用时注册需要序列化的类型，建议在对网络敏感的应用场景下使用Kryo。  
一个例子：

val sparkConf = new SparkConf()
          /*.setMaster("local")*/
          .setAppName("Test")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          /*.set("spark.kryo.registrationRequired", "true")*/
          .registerKryoClasses(Array(
          classOf[Array[String]],
          classOf[util.HashMap[String, String]],
          classOf[util.ArrayList[String]],
    	  classOf[MyTest]
        ))

在集群跑的时候把/\*.set("spark.kryo.registrationRequired", "true")\*/注释掉，否则会报如下错误：

Class is not registered: scala.reflect.ClassTag$$anon$1或者Class is not registered: scala.reflect.ManifestFactory$$anon$9

最后，如果你不注册需要序列化的自定义类型，Kryo也能工作，不过每一个对象实例的序列化结果都会包含一份完整的类名，这很浪费没必要的空间。