什么时候可以序列化 Spark 任务？答案

【问题标题】：When exactly a Spark task can be serialized?什么时候可以序列化 Spark 任务？
【发布时间】：2015-05-10 05:15:33
【问题描述】：

我阅读了有关此主题的一些相关问题，但仍然无法理解以下内容。我有这个简单的 Spark 应用程序，它从文件中读取一些 JSON 记录：

object Main {

  // implicit val formats = DefaultFormats // OK: here it works

  def main(args: Array[String]) {
    val conf = new SparkConf().setMaster("local").setAppName("Spark Test App")
    val sc = new SparkContext(conf)
    val input = sc.textFile("/home/alex/data/person.json")

    implicit val formats = DefaultFormats // Exception: Task not serializable

    val persons = input.flatMap { line ⇒
      // implicit val formats = DefaultFormats // OK: here it also works
      try {
        val json = parse(line)
        Some(json.extract[Person])
      } catch {
        case e: Exception ⇒ None
      }
    }
  }
}

我认为隐含的formats 是不可序列化的，因为它包含一些日期格式的ThreadLocal。但是，为什么它作为object Main 的成员或flatMap 的闭包中的成员而不是main 函数中的常见val 时起作用？

提前致谢。

【问题讨论】：

标签： scala serialization closures apache-spark

【解决方案1】：

如果formats 在flatMap 内，则它只是作为执行映射函数的一部分而创建的。所以映射器可以被序列化并发送到集群，因为它还不包含formats。另一方面，这将在每次映射器运行时重新创建formats（即每行一次）-您可能更喜欢使用mapPartitions 而不是flatMap，这样您就可以为每个分区创建一次值。

如果formats 在flatMap 之外，那么它会在主计算机上创建一次，您正在尝试对其进行序列化并将其发送到集群。

我不明白为什么 formats 作为 Main 的字段会起作用。也许objects 是神奇的伪序列化，因为它们是单例（即它们的字段实际上没有序列化，而是对单个静态Main 实例的引用被序列化）？不过这只是猜测。

【讨论】：

【解决方案2】：

我认为回答您问题的最佳方法是三个简短的答案：

1) 为什么它作为 Main 对象的成员可以工作？，这里的问题是代码可以工作，因为它位于 Object 内部，而不是 Main Object。现在：为什么？因为 Spark 会序列化你的整个对象并将其发送给每个 executor，此外，Scala 中的 Object 就像 JAVA 静态类一样生成，Java 类中静态字段的初始值存储在 jar 中，worker 可以直接使用它。如果您使用类而不是对象，则情况不同。

2) 第二个问题是：为什么它在平面地图中有效？。当您在 RDD（过滤器、flatMap ......等）上运行转换时，您的转换代码是：在驱动程序节点上序列化，发送到工作人员，一旦在那里它将被反序列化并执行。如您所见，与 1) 中的代码完全相同。代码将“自动”序列化。

最后是 3) 问题：为什么这不能作为 main 函数中的公共 val 工作？这是因为 val 不是“自动”序列化的，但您可以像这样测试它：val yourVal = new yourVal with Serializable

【讨论】：