如何让Spark使用Kryo序列化一个对象？

Question

object HelloSpark {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
                .setAppName("Testing HelloSpark")
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .set("spark.kryo.registrator", "xt.HelloKryoRegistrator")

        val sc = new SparkContext(conf)
        val rdd = sc.parallelize(1 to 20, 4)
        val bytes = new ImmutableBytesWritable(Bytes.toBytes("This is a test"))

        rdd.map(x => x.toString + "-" + Bytes.toString(bytes.get) + " !")
            .collect()
            .foreach(println)

        sc.stop
    }
}

// My registrator
class HelloKryoRegistrator extends KryoRegistrator {
    override def registerClasses(kryo: Kryo) = {
        kryo.register(classOf[ImmutableBytesWritable], new HelloSerializer())
    }
}

//My serializer 
class HelloSerializer extends Serializer[ImmutableBytesWritable] {
    override def write(kryo: Kryo, output: Output, obj: ImmutableBytesWritable): Unit = {
        output.writeInt(obj.getLength)
        output.writeInt(obj.getOffset)
        output.writeBytes(obj.get(), obj.getOffset, obj.getLength)
    }

    override def read(kryo: Kryo, input: Input, t: Class[ImmutableBytesWritable]): ImmutableBytesWritable = {
        val length = input.readInt()
        val offset = input.readInt()
        val bytes  = new Array[Byte](length)
        input.read(bytes, offset, length)

        new ImmutableBytesWritable(bytes)
    }
}

null

但是，当我以Yar-Client模式提交Spark应用程序时，抛出了以下异常：

线程“main”org.apache.spark.sparkException：Task not serializable at org.apache.spark.util.closureCleaner$.ensureRecleaner.ccala:166)at org.apache.spark.util.closureCleaner$.clean(closureCleaner.scala:158)at org.apache.spark.sparkcontext.clean(sparkcontext.scala:1242)at org.apache.spark.rdd.rdd.map(rdd.scala:270)at Scala）在sun.reflect.nativeMethodAccessorImpl.invoke0（原生方法）在sun.reflect.nativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)在sun.reflect.delegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)在java.lang.reflect.Method.invoke(Method.java:606)在ksubmit.scala）引起的原因：java.io.NotSerializableException：org.apache.hadoop.hbase.io.immutab在java.io.objectOutputStream.writeObject0(ObjectOutputStream.java:1183)在java.io.objectOutputStream.DefaultWriteFields(ObjectOutputStream.java:1547)在java.io.objectOutputStream.writeRegialData(ObjectOutputStream.writeRegialObjectOutputStream.java:1508)在ectoutputstream.java:347)在org.apache.spark.serializer.javaserializationstream.writeObject(javaserializer.scala:42)在org.apache.spark.serializer.javaserializerinstance.serializer(javaserializer.scala:73)，在org.apache.spark.util.closurecleaner$.ensureResializable(closurecleaner.scala:164)...12个以上

似乎ImmutableBytesWritable不能被Kryo序列化。那么让Spark使用Kryo序列化对象的正确方法是什么呢？Kryo可以序列化任何类型吗？

Answer 1

发生这种情况是因为在闭包中使用了ImmutableBytesWritable。Spark还不支持Kryo的闭包序列化（只支持RDDs中的对象）。你可以利用这个帮助来解决你的问题：

Spark-task不可序列化：如何处理调用外部类/对象的复杂映射闭包？

您只需在传递闭包之前序列化对象，然后反序列化。即使类不能序列化，这种方法也很有效，因为它在幕后使用Kryo。你只需要一些咖喱。；）

def genMapper(kryoWrapper: KryoSerializationWrapper[(Foo => Bar)])
               (foo: Foo) : Bar = {
    kryoWrapper.value.apply(foo)
}
val mapper = genMapper(KryoSerializationWrapper(new ImmutableBytesWritable(Bytes.toBytes("This is a test")))) _
rdd.flatMap(mapper).collectAsMap()

object ImmutableBytesWritable(bytes: Bytes) extends (Foo => Bar) {
    def apply(foo: Foo) : Bar = { //This is the real function }
}