在 Apache Spark 中的分区 RDD 上使用 foreach 时输出不一致：应该避免吗？

Question

分区 RDD 上的 foreach 操作会产生不可预测的结果。为什么会出现这种情况？

例如，我尝试在划分为两个切片 (numSlices=2) 的 RDD 上使用 foreach 操作来打印双倍值。这是代码：

numbers = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], numSlices=2)
print(numbers.collect())

def f(x):
    print(x, x*2)
    
numbers.foreach(lambda x: f(x))

但是，输出不一致。有时它会正确显示由原始值 x 和 2x。但是当我重新运行代码时，输出有所不同，如下所示。在使用分区 RDD 时我应该避免使用 foreach 吗？

[1, 2, 3, 4, 5, 6, 7, 8, 9]
15  102

62  124

73  146

84  168

9 18

Answer 1

Apache Spark 是一个并行计算引擎。它不是顺序处理分区，而是并行处理。让我们分析一下结果的第一行。

print(x, x*2)

基本上按顺序执行以下操作：

print(x)

、

print(' ')

、

print(x*2)

、

print(\n)

。现在，让我们考虑

1

和

5

位于不同的分区中。如果它们由两个单独的执行器并行处理，则可能会发生以下操作序列：

print(1) # executor_1
print(5) # executor_2
print(' ') # executor_1
print(' ') # executor_2
print(5*2) # executor_2
print(1*2) # executor_1
print(\n) # executor_2
print(\n) # executor_1

对于两个执行器来说，动作序列是连贯的，但由于两个序列是纠缠在一起的，所以会产生你所看到的奇怪的输出：

15  102

请注意，如果强制 Spark 在同一核心上处理所有内容，输出会变得更加连贯：

>>> numbers.coalesce(1).foreach(lambda x: f(x))
1 2
2 4
3 6
4 8
5 10
6 12
7 14
8 16
9 18
10 20