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

Question

看看这个问题：Scala Spark-任务不可序列化：java.io.NotSerializableExceptionon。当调用函数外部闭包只对类不是对象。

问题:

假设我的映射器可以是函数（def），在内部调用其他类，创建对象，并在内部执行不同的操作。（或者它们甚至可以是扩展的类（Foo）=

Spark仅支持闭包的Java序列化。有办法解决这个问题吗？我们可以用某物代替闭包来做我想做的事情吗？我们可以很容易地用Hadoop做这类事情。这件事让火花对我来说几乎无法使用。我们不能期望所有的第三方库都具有可扩展的所有类。

可能的解决方案：

像这样的东西有什么用处吗：https://github.com/amplab/shark/blob/master/src/main/scala/shark/execution/serialization/KryoSerializationWrapper.scala

当然，答案似乎是包装纸，但我看不出确切的答案。

Answer 1

在使用JavaAPI的情况下，当传递到映射函数闭包时，应该避免匿名类。您需要一个类来扩展您的函数并将其传递给map（..），而不是执行map（新函数）见：https://yanago.wordpress.com/2015/03/21/apache-spark/

Answer 2

我自己想出了办法！

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

这里有一个我是如何做到的例子：

def genMapper(kryoWrapper: KryoSerializationWrapper[(Foo => Bar)])
               (foo: Foo) : Bar = {
    kryoWrapper.value.apply(foo)
}
val mapper = genMapper(KryoSerializationWrapper(new Blah(abc))) _
rdd.flatMap(mapper).collectAsMap()

object Blah(abc: ABC) extends (Foo => Bar) {
    def apply(foo: Foo) : Bar = { //This is the real function }
}

您可以随意使Blah变得复杂，比如类、伴随对象、嵌套类、对多个第三方lib的引用。

KryoSerializationWrapper指：https://github.com/amplab/shark/blob/master/src/main/scala/shark/execution/serialization/KryoSerializationWrapper.scala