使用匿名函数时的 Spark TaskNotSerializable答案

【问题标题】：Spark TaskNotSerializable when using anonymous function使用匿名函数时的 Spark TaskNotSerializable
【发布时间】：2015-12-29 19:51:22
【问题描述】：

背景

这是我的情况：我正在尝试创建一个基于内容的某些特性过滤 RDD 的类，但该特性在不同的场景中可能会有所不同，所以我想用一个函数对其进行参数化。不幸的是，我似乎遇到了 Scala 捕获其闭包的方式的问题。尽管我的函数是可序列化的，但类不是。

从spark source on closure cleaning 中的示例看来，我的情况似乎无法解决，但我相信有一种方法可以通过创建正确的（较小的）闭包来实现我想要做的事情。

我的代码

class MyFilter(getFeature: Element => String, other: NonSerializable) {
  def filter(rdd: RDD[Element]): RDD[Element] = {
    // All my complicated logic I want to share
    rdd.filter { elem => getFeature(elem) == "myTargetString" }     
}

简化示例

class Foo(f: Int => Double, rdd: RDD[Int]) { 
  def go(data: RDD[Int]) = data.map(f) 
}

val works = new Foo(_.toDouble, otherRdd)
works.go(myRdd).collect() // works

val myMap = Map(1 -> 10d)
val complicatedButSerializableFunc: Int => Double = x => myMap.getOrElse(x, 0)
val doesntWork = new Foo(complicatedButSerializableFunc, otherRdd)
doesntWork.go(myRdd).collect() // craps out

org.apache.spark.SparkException: Task not serializable
Caused by: java.io.NotSerializableException: $iwC$$iwC$Foo
Serialization stack:
    - object not serializable (class: $iwC$$iwC$Foo, value: $iwC$$iwC$Foo@61e33118)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, name: foo, type: class $iwC$$iwC$Foo)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC, $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC@47d6a31a)
    - field (class: $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1, name: $outer, type: class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC)
    - object (class $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$anonfun$1, <function1>)

// Even though
val out = new ObjectOutputStream(new FileOutputStream("test.obj"))
out.writeObject(complicatedButSerializableFunc) // works

问题

为什么第一个简化示例不尝试序列化所有 Foo 而第二个示例却尝试序列化？
如何在不包含对 Foo 的引用的情况下获取对我的可序列化函数的引用？

【问题讨论】：

您是在使用 spark-shell 来测试您的代码还是在 spark 应用程序中？
@eliasah 这是在 spark-shell 中
那么你的 spark 版本呢？
对我来说在 1.2、1.4 和 1.5 中失败。
当然很高兴它有帮助。编辑以消除混淆。

标签： scala apache-spark closures serializable

【解决方案1】：

在this article的帮助下找到了答案。

本质上，当为给定函数创建闭包时，Scala 将包含引用的任何复杂字段的整个对象（如果有人很好地解释了为什么在第一个简单示例中没有发生这种情况，我会接受回答）。解决方案是将可序列化的值传递给不同的函数，以便只保留最小的引用，非常类似于事件侦听器的 ol' javascript for-loop 范例。

示例

def enclose[E, R](enclosed: E)(func: E => R): R = func(enclosed)

class Foo(f: Int => Double, somethingNonserializable: RDD[String]) { 
 def go(data: RDD[Int]) = enclose(f) { actualFunction => data.map(actualFunction) } 
}

或者用JS风格的自执行匿名函数

def go(data: RDD[Int]) = ((actualFunction: Int => Double) => data.map(actualFunction))(f)

【讨论】：