spark reduce 功能：了解它的工作原理

Question

我正在接受这个course。

它表示 RDD 上的 reduce 操作是一次在一台机器上完成的。这意味着如果您的数据分布在两台计算机上，那么下面的函数将处理第一台计算机中的数据，将找到该数据的结果，然后它将从第二台计算机获取单个值，运行该函数，它将继续这样，直到它完成机器 2 中的所有值。这是正确的吗？

我以为该函数会同时在两台机器上开始运行，然后一旦有两台机器的结果，它将再次运行该函数最后一次

rdd1=rdd.reduce(lambda x,y: x+y)

更新 1--------------------------------------------

与减少功能相比，以下步骤会给出更快的答案吗？

Rdd=[3,5,4,7,4]
seqOp = (lambda x, y: x+y)
combOp = (lambda x, y: x+y)
collData.aggregate(0, seqOp, combOp)

更新 2-----------------------------------

下面的两组代码是否应该在相同的时间内执行？我检查了一下，似乎两者都需要相同的时间。

import datetime

data=range(1,1000000000)
distData = sc.parallelize(data,4)
print(datetime.datetime.now())
a=distData.reduce(lambda x,y:x+y)
print(a)
print(datetime.datetime.now())

seqOp = (lambda x, y: x+y)
combOp = (lambda x, y: x+y)
print(datetime.datetime.now())
b=distData.aggregate(0, seqOp, combOp)
print(b)
print(datetime.datetime.now())

Answer 1

reduce 的行为在本机 (Scala) 和来宾语言 (Python) 之间略有不同，但简化了一点：

• 每个分区按元素顺序处理
• 多个分区可以由单个工作人员（多个执行程序线程）或不同工作人员同时处理
• 部分结果被提取到应用最终归约的驱动程序（这是在 PySpark 和 Scala 中具有不同实现的部分）

由于您使用的是 Python，让我们看一下代码：

1. reduce creates a simple wrapper 用于用户提供的功能：

   def func(iterator):
       ...
   

2. 这是包装is used to mapPartitions:

   vals = self.mapPartitions(func).collect()
   

   很明显这段代码是并行的，并不关心结果是如何被利用的

3. 使用标准 Python reduce 在驱动程序上按顺序减少收集的 vals：

   reduce(f, vals)
   

   其中f 是传递给RDD.reduce 的函数

相比之下，Scala 会异步合并来自工作线程的部分结果。

在treeReduce 的情况下，步骤 3. 也可以以分布式方式执行。见Understanding treeReduce() in Spark

总结reduce，不包括驱动程序端处理，使用与map或filter等基本转换完全相同的机制（mapPartitions），并提供相同级别的并行性（再次不包括驱动程序代码）。如果您有大量分区或f 很昂贵，您可以使用tree* 系列方法进行并行/分发最终合并。