我开始使用 Pyspark 进行一些数据处理。我可以做一些像

rdd.map(lambda x : (x['somekey'], 1)).reduceByKey(lambda x,y: x+y).count()

它会将这些函数中的逻辑发送到可能多台机器上以并行执行。

现在，如果我有 Java 背景，如果我想将包含某些方法的对象发送到另一台机器，那台机器需要知道通过网络流式传输的对象的类定义。最近 java 有了函数式接口(interface)的想法，它将在编译时为我创建该接口(interface)的实现(即 MyInterface impl = ()->System.out.println("Stuff");)

MyInterface 只有一种方法，'doStuff()'

但是，如果我想通过线路发送这样的函数，目标机器需要知道实现(impl 本身)才能调用它的“doStuff()”方法。

我的问题归结为...用 Scala 编写的 Spark 如何实际将功能发送到其他机器？我有几个预感:

1. 驱动程序将类定义流式传输到其他机器，这些机器使用类加载器动态加载它们。然后驱动程序流式传输对象，机器知道它们是什么，并可以在它们上执行。
2. Spark 在所有机器(核心库)上定义了一组方法，这些方法是我可以传递给它的任何东西所需要的。也就是说，我传递的函数被转换为对核心库的一个或多个函数调用。 (似乎不太可能，因为 lambda 可以是任何东西，包括在内部实例化其他对象)

谢谢!

编辑:Spark 是用 Scala 编写的，但我很想知道如何在 Java 中实现这一点(其中一个函数只有在一个类中才能存在，因此改变了需要在工作节点上更新的类定义)。

编辑 2: 这是java中遇到混淆的问题:

public class Playground
{
    private static interface DoesThings
    {
        public void doThing();
    }
    public void func() throws Exception {
        Socket s = new Socket("addr", 1234);
        ObjectOutputStream oos = new ObjectOutputStream(s.getOutputStream());
        oos.writeObject("Hello!"); // Works just fine, you're just sending a string
        oos.writeObject((DoesThings)()->System.out.println("Hey, im doing a thing!!")); // Sends the object, but error on other machine

        DoesThings dt = (DoesThings)()->System.out.println("Hey, im doing a thing!!");
        System.out.println(dt.getClass());
    }
}

System.out,println(dt.getClass()) 返回: “类 JohnLibs.Playground$$Lambda$1/23237446”

现在，假设接口(interface)定义不在同一个文件中，而是在两台机器都有的共享文件中。但是这个驱动程序 func() 实质上创建了一种实现 DoesThings 的新型类。 如您所见，目标机器不知道 JohnLibs.Playground$$Lambda$1/23237446 是什么，即使它知道 DoesThings 是什么。这一切都归结为您不能在不绑定(bind)到类的情况下传递函数。在 python 中，您可以只发送一个带有定义的字符串，然后执行该字符串(因为它已被解释)。也许这就是 spark 所做的，因为它使用 scala 而不是 java(如果 scala 可以在类之外具有函数)

最佳答案

Java 字节码，当然是 Java 和 Scala 的编译目标，是专门为平台独立而创建的。因此，如果您有一个类文件，您可以将它移动到任何其他机器，而不管“硅”架构如何，并且只要它具有至少该版本的 JVM，它就会运行。 James Gosling 和他的团队故意这样做是为了让代码从一开始就在机器之间移动，并且在 Java 0.98(我使用的第一个版本)中很容易演示。

当 JVM 试图加载一个类时，它会使用 ClassLoader 的一个实例。类加载器包含两件事，获取字节码文件的二进制文件的能力，以及加载代码的能力(验证其完整性，将其转换为内存中的 java.lang.Class 实例，并使其可用于其他代码在系统中)。在 Java 1 中，如果您想控制 bye 的加载方式，您大多必须编写自己的类加载器，尽管有一个特定于 sun 的 AppletClassLoader，它是为从 http 而不是从文件系统加载类文件而编写的。

稍后，在 Java 1.2 中，“如何获取类文件的字节”部分在 URLClassloader 中被分离出来。这可以使用任何支持的协议(protocol)来加载类。实际上，协议(protocol)支持机制过去和现在都可以通过可插入的协议(protocol)处理程序进行扩展。因此，现在您可以从任何地方加载类，而不会在较难的部分出错，这就是您验证和安装类的方式。

除此之外，Java 的 RMI 机制允许将序列化对象(类名，以及对象的“状态”部分)包装在 MarshaledObject 中。这添加了“可以从哪里加载此类”，表示为 URL。 RMI 自动将内存中的真实对象转换为 MarshaledObjects，并在网络上传送它们。如果 JVM 接收到一个它已经具有类定义的编码对象，它总是使用该类定义(为了安全)。但是，如果没有，那么只要满足一系列标准(安全性，以及简单的正常工作标准)，就可以从该远程服务器加载类文件，从而允许 JVM 加载它从未见过定义的类。 (显然，此类系统的代码通常必须针对无处不在的接口(interface)编写——否则，将会有大量的反射发生!)

现在，我不知道(事实上，我发现你的问题试图确定 Spark 是否使用 RMI 基础设施(我知道 hadoop 没有，因为，似乎是因为作者想创建他们自己的系统- - 这当然是有趣和有教育意义的 - 而不是使用灵活的、可配置的、经过广泛测试的，包括经过安全测试的!- 系统。)

但是，要使这项工作大体上必须发生的就是我为 RMI 概述的步骤，这些要求基本上是:

1) 对象可以序列化为所有参与者都能理解的某种字节序列格式

2) 当通过线路发送对象时，接收端必须有某种方法来获取定义它们的类文件。这可以是 a) 预安装，b) RMI 的“在这里找到这个”的方法或 c) 发送系统发送 jar。这些都可以工作

3) 应该维护安全。在 RMI 中，这个要求相当“在你面前”，但我在 Spark 中看不到它，所以他们要么隐藏了配置，要么只是修复了它可以做什么。

无论如何，这不是一个真正的答案，因为我用一个具体的例子描述了原则，而不是对你的问题的实际具体答案。我还是想找到那个!

关于java - Apache Spark 如何将函数发送到引擎盖下的其他机器，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/42096854/