Spark中的ReduceByKey函数答案

【问题标题】：ReduceByKey function In SparkSpark中的ReduceByKey函数
【发布时间】：2016-02-25 19:54:45
【问题描述】：

我在某处读到，对于作用于单个 RDD 的操作，例如 reduceByKey()，在预分区 RDD 上运行将导致每个键的所有值都在单个机器上本地计算，只需要从每个工作节点发送回主节点的最终局部缩减值。这意味着我必须声明一个分区器，例如：

val sc = new SparkContext(...)
val userData = sc.sequenceFile[UserID, UserInfo]("hdfs://...")
             .partitionBy(new HashPartitioner(100))   // Create 100 partitions
             .persist()

为了让reduceByKey 像我之前解释的那样工作。

我的问题是，如果我想使用 reduceByKey（最佳），我是否需要每次都声明一个分区器，还是没有必要。

【问题讨论】：

标签： scala apache-spark rdd partitioning reduce

【解决方案1】：

其实你说的这两种品质有些不相干。

对于reduceByKey()，第一质量使用提供的关联reduce 函数本地首先在每个执行器上聚合相同键的元素，然后最终跨执行器聚合。它被封装在一个名为 mapSideCombine 的布尔参数中，如果设置为 true，则执行上述操作。如果设置为 false，就像 groupByKey() 一样，每条记录都将被打乱并发送到正确的执行者。

第二个质量涉及分区及其使用方式。每个 RDD，根据其定义，包含一个拆分列表和（可选）一个分区器。方法reduceByKey() 被重载，实际上有一些定义。例如：

def reduceByKey(func: (V, V) => V): RDD[(K, V)]

该方法的定义实际上使用了来自父RDD的默认现有分区器，并减少到设置为默认并行度级别的分区数。
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]

该方法的定义将使用HashPartitioner 将数据分配给其相应的执行程序，并且分区数将为numPartitions。
def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]

最后，该方法的定义取代了其他两个，并采用了一个通用的（也许是自定义的）分区器，该分区器将产生由分区器如何对键进行分区决定的分区数量。

关键是您实际上可以在reduceByKey() 本身内编码您想要的分区器逻辑。如果您的意图是通过预分区来避免改组开销，那么这也没有任何意义，因为您仍将在预分区上进行改组。

【讨论】：

【解决方案2】：

在执行reduceByKey 时对RDD 进行分区以避免网络流量并不是最佳解决方案。即使它不需要对reduceByKey 进行洗牌，它也必须洗牌一个完整的数据集才能执行分区。

由于这通常要昂贵得多，因此预分区没有意义，除非您的目标是以增加整体延迟为代价来减少 reduceByKey 阶段的延迟，或者您可以将此分区用于其他任务。

【讨论】：

【解决方案3】：

不是真的。 reduceByKey 正在使用数据局部性。来自 RDD api：

/** * 使用关联归约函数合并每个键的值。这也将执行 * 在将结果发送到 reducer 之前，在每个 mapper 上进行本地合并，类似于 a * MapReduce 中的“组合器”。 */

这意味着当您有一个键值 RDD 时，在第一阶段，使用提供的函数对每个分区级别的相同键进行归约，然后使用相同的函数对所有已聚合值。无需提供分区程序。它只是工作。

【讨论】：