在Spark中需要kryo序列化(Scala)

Question

我打开了kryo序列化:

conf.set( "spark.serializer", "org.apache.spark.serializer.KryoSerializer" )

我想确保在节点之间进行混洗时使用kryo序列化自定义类.我可以这样用kryo注册这个类:

conf.registerKryoClasses(Array(classOf[Foo]))

据我了解,这实际上并不能保证使用kyro序列化; 如果序列化程序不可用,kryo将回退到Java序列化.

为了保证kryo序列化的发生,我遵循了Spark文档中的这个建议:

conf.set("spark.kryo.registrationRequired", "true")

但这会导致抛出IllegalArugmentException("Class未注册"),因为我认为Spark会在内部使用一堆不同的类,例如:

org.apache.spark.util.collection.CompactBuffer
scala.Tuple3

当然,我不必用kryo手动注册每个单独的类？这些序列化程序都是用kryo定义的,那么有没有办法自动注册所有这些序列化程序？

Answer 1

据我了解,这实际上并不能保证使用kyro序列化; 如果序列化程序不可用,kryo将回退到Java序列化.

不.如果你设置spark.serializer为org.apache.spark.serializer. KryoSerializer那么Spark将使用Kryo.如果Kryo不可用,您将收到错误消息.没有后退.

那么这个Kryo注册是什么呢？

当Kryo序列化未注册类的实例时,它必须输出完全限定的类名.这是很多人物.相反,如果一个类已经预先注册,Kryo只能输出一个数字引用到这个类,这只是1-2个字节.

当使用Kryo序列化RDD的每一行时,这尤其重要.您不希望为每十亿行包含相同的类名.所以你预先注册这些类.但是很容易忘记注册一个新类,然后再次浪费字节.解决方案是要求每个类都要注册:

conf.set("spark.kryo.registrationRequired", "true")

现在Kryo永远不会输出完整的班级名称.如果遇到未注册的类,那就是运行时错误.

不幸的是,很难枚举您将要提前序列化的所有类.我们的想法是Spark注册特定于Spark的类,然后注册其他所有类.你有RDD[(X, Y, Z)]吗？你必须注册classOf[scala.Tuple3[_, _, _]].

该的火花登记类列表实际上包含了CompactBuffer,所以如果你看到一个错误,你做错了什么.您正在绕过Spark注册过程.您必须使用spark.kryo.classesToRegister或spark.kryo.registrator注册您的课程.(请参阅配置选项.如果使用GraphX,则您的registrator应调用GraphXUtils.registerKryoClasses.)