Spark：reduce和reduceByKey之间的语义差异答案

【问题标题】：Spark: difference of semantics between reduce and reduceByKeySpark：reduce和reduceByKey之间的语义差异
【发布时间】：2016-05-14 07:20:54
【问题描述】：

在 Spark 的文档中，它说 RDDs 方法 reduce 需要关联 AND 可交换二进制函数。

但是，reduceByKey 方法只需要关联二元函数。

sc.textFile("file4kB", 4)

我做了一些测试，显然这是我得到的行为。为什么会有这种差异？为什么reduceByKey 确保二进制函数始终以特定顺序应用（以适应缺乏交换性）而reduce 不这样做？

例如，如果加载一些具有 4 个分区（最少）的（小）文本：

val r = sc.textFile("file4k", 4)

然后：

r.reduce(_ + _)

返回一个字符串，其中部分的顺序并不总是相同，而：

r.map(x => (1,x)).reduceByKey(_ + _).first

总是返回相同的字符串（其中所有内容的顺序与原始文件中的顺序相同）。

（我检查了r.glom，文件内容确实分布在4个分区，没有空分区）。

【问题讨论】：

我猜reduceByKey 的想法是您可能有很多不同的键，因此可以在单个线程上减少单个键的所有内容，这意味着您始终可以运行计算左到右。相比之下，reduce 经常用于大型数据集，因此不必关心操作顺序。
您在实验中使用了多少个执行器？

【解决方案1】：

根据代码文档，最近更新/更正。（感谢@zero323）：

reduceByKey 使用关联和交换归约函数合并每个键的值。这还将在将结果发送到减速器之前在每个映射器上本地执行合并，类似于 MapReduce 中的“组合器”。

所以这实际上是像@zero323 在他的回答中指出的文档错误。

您可以检查以下代码链接以确保：

【讨论】：

【解决方案2】：

就我而言，这是文档中的错误，您看到的结果只是偶然的。实践，other resources 和一个简单的analysis of the code 表明传递给reduceByKey 的函数不仅应该是关联的，而且应该是可交换的。

实践 - 虽然看起来订单在本地模式下保留，但在集群上运行 Spark 时不再正确，包括独立模式。
其他资源 - 从AmpCamp 3 引用Data Exploration Using Spark：

Spark 中有一个名为 reduceByKey 的便捷方法正是针对这种模式。请注意，reduceByKey 的第二个参数决定了要使用的 reducer 的数量。默认情况下，Spark 假定 reduce 函数是可交换和关联的，并在映射器端应用组合器。
代码 - reduceByKey 使用 combineByKeyWithClassTag 实现并创建 ShuffledRDD。由于 Spark 不保证改组后的顺序，因此恢复它的唯一方法是将一些元数据附加到部分减少的记录中。据我所知，没有这样的事情发生。

在旁注reduce 因为它在 PySpark 中实现，它可以与仅可交换的函数一起正常工作。当然，这只是实现的一个细节，而不是合同的一部分。

【讨论】：

我要补充一点，reduce 是一个动作，将数据返回给驱动程序，而 reduceByKey 是一个转换，返回另一个 RDD
谢谢！但是，Spark 中是否有某种方法可以确保非交换处理的正确性？还是超出了 Spark 的范围？
我不确定是否理解这个问题。你问是否可以自动测试/证明可交换性，或者只是想用reduce 使用非交换函数？如果这是模仿 PySpark 行为的第二种情况 (mapPartitions(reduceFunc) => collect => reduce(reduceFunc)`) 应该会带来一些性能损失。
感谢您的提示。你的意思是 PySpark 的 reduce 有一个有序的行为？为什么会有这种差异？
附带问题：您似乎在诉诸 collect 返回一个包含原始顺序分区的数组。这也是我观察到的一种行为，但没有记录在案。这是一种可靠的行为吗？