原始 java 代码在 Spark 中究竟执行在哪里？答案

【问题标题】：Where exactly the raw java code execute in Spark?原始 java 代码在 Spark 中究竟执行在哪里？
【发布时间】：2021-06-22 04:07:46
【问题描述】：

我是 Spark 的新手，我知道 Spark 通常会序列化函数并将其发送给所有执行程序，然后处理 HDFS 中可用的数据块。但是如果我有以下代码，

Random random = new Random(); //statement A
int randomValue = random.nextInt(); //statement B
JavaPairRDD<String, Integer> pairRDD = mapRandom.mapToPair(s -> {
    return new Tuple2<>(s, randomValue);
});

A, B 语句到底在哪里运行？当然，它不会在每个执行器中执行它，因为每个执行器都运行自己的 JVM，我发现每个值都映射到完全相同的randomValue。

这是否意味着这些语句A、B 在驱动程序中运行并被复制到所有执行程序？

【问题讨论】：

标签： java apache-spark rdd

【解决方案1】：

A 和 B 在驱动程序上运行并发送给执行程序。

如果你想在 executors 中运行它，你应该这样做：

JavaPairRDD<String, Integer> pairRDD = mapRandom.mapToPair(s -> {
    Random random = new Random(); //statement A
    int randomValue = random.nextInt(); //statement B
    return new Tuple2<>(s, randomValue);
});

【讨论】：

【解决方案2】：

RDD 是分布式的（因此得名），但它是在驱动程序代码中定义的，因此任何不与 RDD 内容交互的代码都是在驱动程序上执行的代码（根据经验），而任何具有与 RDD 内容相关的内容将在 executors 上运行。

如您所见，驱动程序计算您的 randomValue 一次并将其发送给所有执行程序，因为 mapToPair lambda 参数关闭了该计算值。

在 lambda 中移动（仅）random.nextInt() 调用将触发对分布式集合中的每个元素执行该调用。此外，random 值本身将被序列化并通过网络发送。

在 lambda 中移动随机创建本身会使其更小（没有要捕获的外部状态），但会为分布式集合中的每个元素创建一个新的 Random 实例，这显然不是最佳的。

每个执行器要有一个随机实例值，您可以将其设为静态成员（或惰性单例），每个 JVM/执行器初始化一次。要为每个（比如说）分区设置不同的随机值，您可能应该使用 forEachPartition 或类似名称，使用 nextInt() 生成一个新值，并将该值用于该分区中的所有元素。

查看更高级别的 DataFrame/SQL API，因为您可能会发现实现所需内容要容易得多，甚至无需担心代码在哪里执行。

【讨论】：