Spark - 为什么在打印 RDD 之前需要收集（）到驱动程序节点？不能并行吗？

Question

我正在阅读有关如何在 Spark 中打印 RDD（我正在使用 Java）的信息，似乎大多数人只是 collect()（如果 RDD 足够小）并使用 forall(println) 或类似的东西.不能并行打印吗？为什么我们必须将数据收集到驱动节点上才能打印？

我在想也许是因为我们不能并行使用 System.out，但我觉得不是这样。此外，就代码而言，我不太确定如何分配数据并并行打印。我想到的一种方法是做一个 mappartitions，它在映射方面没有做任何有用的事情，但它会遍历分区并打印其内容。

Answer 1

当您调用collect() 方法时，您会将所有结果返回给驱动程序节点。您将拥有List 而不是RDD。让我们看一个本地模式的例子。假设你有一个整数 RDD：

JavaRDD<Integer> rdd = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));

如果您调用foreach 方法（Java 中的stream().forEach()），驱动程序节点将按照您创建它的顺序打印RDD 中的所有元素。

rdd.collect().stream().forEach(x -> System.out.println(x));

Output: 1, 2, 3, 4, 5, 6, 7, 8, 9, 10

如果你想在每个worker上打印结果，你必须调用RDD的foreach方法。它不会向驱动程序返回任何内容，只会在每个工作节点上执行您在 foreach 方法中指定的计算。

rdd.foreach(x -> System.out.println(x));

如果您看到控制台（本地模式），您会注意到 System.out.println(x) 已在单独的线程中执行，因为输出不遵守原始顺序：

Output: 6, 3, 2, 1, 8, 9, 10, 4, 5, 7

所以如果你以分布式模式执行它，每个执行器都会在其日志文件上打印System.out.println 操作的结果。

您还提到了mapPartitions 方法。在您的情况下，我发现它没有比直接在RDD 上使用foreach 更有用。这可能有助于控制工人的工作量。

 rdd.repartition(5).mapPartitions(x -> {
     while(x.hasNext()){
         Integer i = x.next();
         System.out.println(i);
     }
     return x;
 }).count(); // Count is just to force the execution of mapPartition (mapPartition is lazy and doesn't get executed until an action is called)

希望对你有帮助！