java - Spark - 任务不可序列化 : How to work with complex map closures that call outside classes/objects?

Question

看看这个问题:Scala + Spark - Task not serializable: java.io.NotSerializableExceptionon. When calling function outside closure only on classes not objects .

问题:

假设我的映射器可以是内部调用其他类并创建对象并在内部执行不同操作的函数(def)。 (或者它们甚至可以是扩展 (Foo) => Bar 的类并在它们的 apply 方法中进行处理 - 但现在让我们忽略这种情况)

Spark 仅支持闭包的 Java 序列化。有没有办法解决这个问题？我们可以使用一些东西而不是闭包来做我想做的事吗？我们可以使用 Hadoop 轻松完成此类工作。这件事让我几乎无法使用 Spark。不能指望所有 3rd 方库的所有类都扩展 Serializable!

可能的解决方案:

这样的东西似乎有用吗:https://github.com/amplab/shark/blob/master/src/main/scala/shark/execution/serialization/KryoSerializationWrapper.scala

答案似乎肯定是包装器，但我不知 Prop 体是怎么做的。

Answer 1

我自己想出了如何做到这一点!

您只需要在通过闭包之前序列化对象，然后再进行反序列化。即使您的类不可序列化，这种方法也很有效，因为它在幕后使用了 Kryo。你只需要一些 curry 。 ;)

这是我如何做到的一个例子:

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 变得尽可能复杂，类、伴生对象、嵌套类、对多个 3rd 方库的引用。

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