scala - 在 Spark 中，在所有工作人员上拥有静态对象的正确方法是什么？

Question

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

Spark’s API relies heavily on passing functions in the driver program to run on the cluster. There are two recommended ways to do this:

Anonymous function syntax, which can be used for short pieces of code. Static methods in a global singleton object. For example, you can define object MyFunctions and then pass MyFunctions.func1, as follows:

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

myRdd.map(MyFunctions.func1) 

Note that while it is also possible to pass a reference to a method in a class instance (as opposed to a singleton object), this requires sending the object that contains that class along with the method. For example, consider:

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

Here, if we create a new MyClass and call doStuff on it, the map inside there references the func1 method of that MyClass instance, so the whole object needs to be sent to the cluster. It is similar to writing 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 如何生成代码的问题。请记住，Scala object几乎是一个充满静态方法的 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。

in this blog 解释了一种在您需要清理时可能有用的方法，在“设置()和清理()”部分。

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

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()方法。