在Spark中，在所有工作线程上拥有静态对象的正确方法是什么？

Question

我一直在查看Spark的文档，其中提到了这一点：

Spark的API在很大程度上依赖于在驱动程序中传递函数以在集群上运行。有两种推荐的方法可以做到这一点：

匿名函数语法，可用于简短的代码段。全局单例对象中的静态方法。例如，您可以定义对象MyFunctions，然后传递MyFunctions.func1，如下所示：

object MyFunctions {   def func1(s: String): String = { ... } }

myRdd.map(MyFunctions.func1) 

请注意，虽然也可以在类实例中传递对方法的引用（与单例对象相对），但这需要将包含该类的对象与方法一起发送。例如，考虑：

class MyClass {   
  def func1(s: String): String = { ... }   
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) } 
} 

在这里，如果我们创建一个新的MyClass并在其上调用doStuff，则其中的映射将引用该MyClass实例的func1方法，因此需要将整个对象发送到集群。它类似于写作 rdd.map(x => this.func1(x))。

现在我的疑问是，如果您在单例对象上具有属性（应该等同于静态属性），将会发生什么。稍作改动的相同示例：

object MyClass {   
  val value = 1   
  def func1(s: String): String = { s + value }   
} 

myRdd.map(MyClass.func1) 

因此，该函数仍然是静态引用的，但是Spark通过尝试序列化所有引用的变量又走了多远？它会序列化value还是在远程工作进程中再次初始化？

此外，在所有情况下，我在一个单例对象中都有一些重型模型，我想找到一种正确的方法将其序列化为工作者，同时保持从任何地方从单例中引用它们的能力，而不是像这样将它们传递给其他人。相当深的函数调用堆栈中的函数参数。

任何有关Spark序列化方式/方式/时间的深入信息将不胜感激。

Answer 1

这不是关于Spark的问题，而是关于Scala如何生成代码的问题。请记住，Scalaobject几乎是一个充满静态方法的Java类。考虑一个像这样的简单示例：

object foo {

  val value = 42

  def func(i: Int): Int = i + value

  def main(args: Array[String]): Unit = {
    println(Seq(1, 2, 3).map(func).sum)
  }

}

那将被翻译成3个Java类。其中之一将是闭包，该闭包是map方法的参数。javap在该类上使用将产生以下内容：

public final class foo$$anonfun$main$1 extends scala.runtime.AbstractFunction1$mcII$sp implements scala.Serializable {
  public static final long serialVersionUID;
  public final int apply(int);
  public int apply$mcII$sp(int);
  public final java.lang.Object apply(java.lang.Object);
  public foo$$anonfun$main$1();
}

请注意，没有任何字段。如果您查看反汇编的字节码，它所做的就是调用func()方法。在Spark中运行时，这是将被序列化的实例；由于它没有字段，因此不需要序列化。

至于您的问题，即如何初始化静态对象，您可以在关闭开始时调用一个幂等的初始化函数。第一个将触发初始化，随后的调用将成为无操作。不过，清理工作要复杂得多，因为我不熟悉执行“在所有执行程序上运行此代码”之类的API。

如果需要清理，一种有用的方法在此博客的“ setup（）和cleanup（）”部分中进行了说明。

编辑：只是为了澄清，这是实际进行调用的方法的反汇编。

public int apply$mcII$sp(int);
  Code:
   0:   getstatic       #29; //Field foo$.MODULE$:Lfoo$;
   3:   iload_1
   4:   invokevirtual   #32; //Method foo$.func:(I)I
   7:   ireturn

了解它如何仅引用保存单例的静态字段并调用该func()方法。