scala - 了解 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

Could someone help me to understand why this might be?

在您的第一个片段中， helperObject是在 run 中声明的局部变量.因此，它将被函数关闭(解除)，这样无论在何处执行此代码，所有信息都将可用，因此 Sparks ClosureCleaner对你大喊大叫，因为你试图序列化它。

在您的第二个代码段中，该值不再是方法范围内的局部变量，它是类实例的一部分(从技术上讲，这是一个对象声明，但它毕竟将被转换为 JVM 类)。

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

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