了解 Apache Spark RDD 任务序列化

Question

我正在尝试了解任务序列化在 Spark 中的工作原理，并且对我在编写的测试中得到的一些混合结果感到有些困惑。

我有一些测试代码（为了发布而简化）在多个节点上执行以下操作：

object TestJob {
  def run(): Unit = {
    val rdd = ...
    val helperObject = new Helper() // Helper does NOT impl Serializable and is a vanilla class
    rdd.map(element => {
      helperObject.transform(element)
    }).collect()
  }
}

当我执行run() 时，由于helperObject 不可序列化，因此作业如预期的那样以“不可序列化的任务”异常发生。但是，当我稍微改变它时，就像这样：

trait HelperComponent {
  val helperObject = new Helper()
}

object TestJob extends HelperComponent {
  def run(): Unit = {
    val rdd = ...
    rdd.map(element => {
      helperObject.transform(element)
    }).collect()
  }
}

作业由于某种原因成功执行。有人可以帮我理解为什么会这样吗？在上述每种情况下，Spark 将序列化并发送给工作人员的具体内容是什么？

我使用的是 Spark 2.1.1 版。

谢谢！

Answer 1

有人可以帮我理解为什么会这样吗？

在您的第一个 sn-p 中，helperObject 是在 run 中声明的局部变量。因此，它将被函数关闭（提升），这样无论何时执行此代码，所有信息都将可用，因此 Sparks ClosureCleaner 会因为您尝试对其进行序列化而大喊大叫。

在你的第二个 sn-p 中，值不再是方法范围内的局部变量，它是 类实例的一部分（从技术上讲，这是一个对象声明，但它会被转换毕竟进入一个JVM类）。

这在 Spark 中很有意义，因为集群中的所有工作节点都包含执行代码所需的 JAR。因此，当 Spark 在您的一个工作人员中启动 Executor 进程时，它不会通过 ClassLoader 在本地加载 TestObject，并创建它的一个实例，而不是为 rdd.map 序列化整个 TestObject，并创建它的实例就像非分布式应用程序中的所有其他 JVM 类一样。

总而言之，您没有看到这种情况发生的原因是由于您声明类型实例的方式发生了变化，该类不再序列化。