任务不可序列化：仅在类而非对象上调用闭包外部函数时，java.io.NotSerializableExcture

Question

在闭包外调用函数时出现奇怪行为：

• 当函数在一个对象中时，一切都在运行
• 当函数位于类get中时：

任务不可序列化：java。伊奥。NotSerializableException:测试

问题是我需要我的代码在一个类中，而不是一个对象。知道为什么会发生这种情况吗？Scala对象是序列化的吗？）？

这是一个工作代码示例：

object working extends App {
    val list = List(1,2,3)

    val rddList = Spark.ctx.parallelize(list)
    //calling function outside closure 
    val after = rddList.map(someFunc(_))

    def someFunc(a:Int)  = a+1

    after.collect().map(println(_))
}

这是一个不起作用的例子：

object NOTworking extends App {
  new testing().doIT
}

//adding extends Serializable wont help
class testing {  
  val list = List(1,2,3)  
  val rddList = Spark.ctx.parallelize(list)

  def doIT =  {
    //again calling the fucntion someFunc 
    val after = rddList.map(someFunc(_))
    //this will crash (spark lazy)
    after.collect().map(println(_))
  }

  def someFunc(a:Int) = a+1
}

Answer 1

完整的谈话充分解释了这个问题，它提出了一个伟大的范式转换方式来避免这些序列化问题：https://github.com/samthebest/dump/blob/master/sams-scala-tutorial/serialization-exceptions-and-memory-leaks-no-ws.md

得票最多的答案基本上是建议放弃整个语言特性——不再使用方法，而只使用函数。的确，在函数式编程中，应该避免类中的方法，但将它们转化为函数并不能解决这里的设计问题（参见上面的链接）。

作为这种特殊情况下的快速修复方法，您可以使用@transient注释告诉它不要尝试序列化有问题的值（这里，Spark.ctx是一个自定义类，而不是Spark在OP的命名之后的类）：

@transient
val rddList = Spark.ctx.parallelize(list)

您还可以重新构造代码，使rddList位于其他地方，但这也很糟糕。

在未来的Scala将包括这些被称为“孢子”的东西，这将使我们能够精确地控制关闭时会和不会被拉进来的东西。此外，这应该将所有意外引入不可序列化类型（或任何不需要的值）的错误变成编译错误，而不是现在，这是可怕的运行时异常/内存泄漏。

http://docs.scala-lang.org/sips/pending/spores.html

当使用kyro，使其注册是必要的，这将意味着你得到的错误，而不是内存泄漏：

“最后，我知道kryo有kryo.setRegistrationOptional（true），但我很难想出如何使用它。当这个选项打开时，如果我没有注册类，kryo似乎仍然会抛出异常。”

向kryo注册课程的策略

当然，这只给您类型级控制，而不是值级控制。

... 还有更多想法。

Answer 2

Grega的回答很好地解释了为什么原始代码不起作用，以及解决问题的两种方法。然而，这种解决方案不是很灵活；考虑一个情况，其中您的闭包包含一个方法调用，在非代码>可序列化的类中，您无法控制。您既不能将Serializable标记添加到此类，也不能更改底层实现以将方法更改为函数。

Nilesh为此提供了一个很好的解决方案，但解决方案可以更加简洁和通用：

def genMapper[A, B](f: A => B): A => B = {
  val locker = com.twitter.chill.MeatLocker(f)
  x => locker.get.apply(x)
}

然后可以使用此函数序列化程序自动包装闭包和方法调用：

rdd map genMapper(someFunc)

这种技术的另一个好处是，访问KryoSerializationWrapper不需要额外的Shark依赖项，因为Twitter的寒意已经被core Spark所吸引

Answer 3

RDD扩展了可序列化接口，因此这并不是导致任务失败的原因。这并不意味着可以使用Spark序列化RDD，避免NotSerializableException

Spark是一个分布式计算引擎，它的主要抽象是一个弹性分布式数据集（RDD），它可以被视为一个分布式集合。基本上，RDD的元素在集群的节点上分区，但是Spark将其抽象为远离用户，让用户与RDD（集合）交互，就像它是本地的一样。

不想涉及太多细节，但当您在RDD上运行不同的转换时（map，flatMap，filter等），您的转换代码（闭包）是：

1. 在驱动节点上序列化，
2. 发送到集群中的适当节点，
3. 反序列化，
4. 并最终在节点上执行

当然，您可以在本地运行（如您的示例中所示），但所有这些阶段（除了通过网络传送）仍然会发生。[这可以让您在部署到生产环境之前就发现任何漏洞]

在第二种情况下，您将调用一个方法，该方法是在map函数内部的类测试中定义的。Spark看到了这一点，因为方法本身无法序列化，Spark试图序列化整个测试类，这样代码在另一个JVM中执行时仍然有效。你有两种可能：

要么让类测试可序列化，这样整个类就可以通过Spark序列化：

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
  val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local[*]"))
}

object NOTworking extends App {
  new Test().doIT
}

class Test extends java.io.Serializable {
  val rddList = Spark.ctx.parallelize(List(1,2,3))

  def doIT() =  {
    val after = rddList.map(someFunc)
    after.collect().foreach(println)
  }

  def someFunc(a: Int) = a + 1
}

或者使用someFunc函数代替方法（函数是Scala中的对象），这样Spark就可以序列化它：

import org.apache.spark.{SparkContext,SparkConf}

object Spark {
  val ctx = new SparkContext(new SparkConf().setAppName("test").setMaster("local[*]"))
}

object NOTworking extends App {
  new Test().doIT
}

class Test {
  val rddList = Spark.ctx.parallelize(List(1,2,3))

  def doIT() =  {
    val after = rddList.map(someFunc)
    after.collect().foreach(println)
  }

  val someFunc = (a: Int) => a + 1
}

类似但不同的类序列化问题可能会引起您的兴趣，您可以在Spark Summit 2013演示中阅读。

顺便说一下，您可以将rddList.map（某些Func（_））重写为rddList.map（某些Func），它们是完全相同的。通常，第二种是首选的，因为它不那么冗长，阅读起来更干净。

编辑（2015-03-15）：SPARK-5307引入了SerializationDebugger，SPARK 1.3.0是第一个使用它的版本。它将序列化路径添加到NotSerializableException。当遇到NotSerializableException时，调试器会访问对象图以查找无法序列化的对象的路径，并构造信息以帮助用户查找对象。

在OP的例子中，这是打印到stdout的内容：

Serialization stack:
    - object not serializable (class: testing, value: testing@2dfe2f00)
    - field (class: testing$$anonfun$1, name: $outer, type: class testing)
    - object (class testing$$anonfun$1, <function1>)