为什么 spark-shell 会吐出 NotSerializableException 而 sbt 控制台不会？

Question

只有一个NotSerializableException 我看不懂。使用 Spark 1.5.0 启动 EMR 集群，然后输入 shell spark-shell --master local（如果您选择 yarn-client 而不是 local，则没有区别）。

然后运行以下代码：

class MaybeSerializable() {
  def add(i: Int) = i + 1
}

val foo = new MaybeSerializable()

sc.makeRDD(1 to 1000).map(foo.add).reduce(_ + _)

我向你保证这会导致org.apache.spark.SparkException: Task not serializable

现在使用相同版本的Spark"org.apache.spark" %% "spark-core" % "1.5.0"创建一个sbt项目，输入sbt console，然后运行

import org.apache.spark.SparkContext

val sc = new SparkContext("local", "shell")

class MaybeSerializable() {
  def add(i: Int) = i + 1
}

val foo = new MaybeSerializable()

sc.makeRDD(1 to 1000).map(foo.add).reduce(_ + _)

我向你保证它会起作用的！也不例外！

如果有人能解释一下，我会很高兴的。

Answer 1

关于行为的不一致，我不是 Spark 专家，但我想说的是，在第一种情况下，当使用 Yarn EMR 时，整个“MaybeSerializable”类肯定会被序列化并发送给执行者由司机。由于它不可序列化，因此会出现“NotSerializableException”异常。

另一方面，在第二种情况下，我不太确定该类是否正在序列化，这就是您没有收到异常的原因。底层架构正在推动或不推动序列化的需求。当您使用 sbt 时，如果驱动程序没有将类的序列化版本发送到任何进程，我不会感到惊讶。在这种情况下，由于不需要它，因此该类是否可序列化并不重要。

我可以在我的计算机上用独立的方式重现第一个问题，然后我摆脱了它，使类以这种方式可序列化：“类 MaybeSerializable() extends Serializable”。这是我使用（和/或匿名函数）的最佳实践，以确保我的代码将在任何集群类型中运行并且序列化行为将在集群之间保持一致。

希望对你有帮助。