火花工之间的代码共享

Question

根据Spark文档，最好在RDD转换中使用匿名或scala对象函数。我有下一个代码的对象：

object Util {
  val someManager = new Manager()

  def process(data: String) = someManager.manage(data)
}

我把它称为下一种方式：

myRDD.map(Util.process)

如何序列化Util对象并将其发送给火花工作人员？经理是为每次发送还是只发送一次而创建的？多少次经理的实例将被发送给火花工人？

Answer 1

如何序列化Util对象并将其发送给火花工作人员？经理是为每次发送还是只发送一次而创建的？多少次经理的实例将被发送给火花工人？

Util.process在编译时被包装在一个AbstractFunction1中，它实现了它的apply方法，这就是被序列化的方法。您可以在解压缩此代码时看到这一点(这是根据JVM字节码重构的)：

par.map(new AbstractFunction1() {
  public static final long serialVersionUID = 0L;

  public final void apply(String data)
  {
    Util..MODULE$.process(data);
  }
 }, ClassTag..MODULE$.Unit());
}

我们希望看到的另一件事是如何将Util对象编译成字节码：

public final class Util$
{
  public static final  MODULE$;
  private final Manager someManager;

  public Manager someManager()
  {
    return this.someManager;
  }

  private Util$()
  {
    MODULE$ = this;
    this.someManager = new Manager();
  }

  public void process(String data)
  {
    someManager().manage(data);
  }
}

这意味着我们仍然有一个用于Manager的实例字段。将会发生的情况是，Spark将使用它的ClosureCleaner来清除Util实例，因为它不需要它，但它将使用serialize将Manager实例序列化给工作人员，并且每次它需要调用map时都会这样做。这并不意味着会有多个Manager实例，它仅仅意味着需要序列化单个实例并通过线路发送。我们可以在ClosureCleaner文档中看到详细信息

• 其机制是遍历封闭闭包的层次结构，并在此过程中空出启动闭包实际使用的任何引用，但这些引用仍然包含在编译的匿名类中。注意到，简单地更改封闭闭包是不安全的，因为其他代码路径可能依赖于它们。相反，我们克隆每个封闭闭包并相应地设置父指针。

Answer 2

Util需要是可序列化的，否则您将得到一个异常。经理每一次都会被派去(如带着每项任务)。不过，每个分区只能发送一次，为此您需要首先广播它，请参阅http://spark.apache.org/docs/latest/programming-guide.html