使用 ReduceByKey 对值列表进行分组答案

【问题标题】：Using ReduceByKey to group list of values使用 ReduceByKey 对值列表进行分组
【发布时间】：2016-06-01 23:17:44
【问题描述】：

我想对每个键的值列表进行分组，并且正在做这样的事情：

sc.parallelize(Array(("red", "zero"), ("yellow", "one"), ("red", "two"))).groupByKey().collect.foreach(println)

(red,CompactBuffer(zero, two))
(yellow,CompactBuffer(one))

但我注意到 Databricks 的一篇博文，它建议不要将 groupByKey 用于大型数据集。

Avoid GroupByKey

有没有办法使用 reduceByKey 实现相同的结果？

我试过了，但它连接了所有值。顺便说一句，就我而言，键和值都是字符串类型。

sc.parallelize(Array(("red", "zero"), ("yellow", "one"), ("red", "two"))).reduceByKey(_ ++ _).collect.foreach(println)

(red,zerotwo)
(yellow,one)

【问题讨论】：

标签： scala hadoop apache-spark mapreduce apache-spark-sql

【解决方案1】：

使用aggregateByKey：

 sc.parallelize(Array(("red", "zero"), ("yellow", "one"), ("red", "two")))
.aggregateByKey(ListBuffer.empty[String])(
        (numList, num) => {numList += num; numList},
         (numList1, numList2) => {numList1.appendAll(numList2); numList1})
.mapValues(_.toList)
.collect()

scala> Array[(String, List[String])] = Array((yellow,List(one)), (red,List(zero, two)))

有关aggregateByKey 的详细信息，请参阅this answer，有关使用可变数据集ListBuffer 背后的基本原理this link。

编辑：

Is there a way to achieve the same result using reduceByKey?

上面的其实性能更差，具体请看@zero323的cmets。

【讨论】：

谢谢@axiom，感激不尽！
与groupByKey 相比，它没有提供任何性能改进。由于 GC 的压力较高，它的效率实际上要低得多。
@zero323 已添加编辑，但这不是密钥分配的功能吗？此外，groupByKey 提供的 Iterable 在本质上是否与 MR 世界的 reducer 中可用的 Iterable 相似（对元素流的抽象，元素可能不在内存中）？。
好的，所以这里有一些问题。对于任何 (key, values) 对，所有元素都必须适合内存。在您的解决方案和 groupByKey 的情况下都是如此。与 groupByKey 不同，您的解决方案应用地图端组合，它不会减少流量，但需要创建本地哈希图来存储对。 Spark 在 groupByKey 中明确避免了这种情况（请参阅 stackoverflow.com/q/33221713/1560062 中由 Climbage 链接的 JIRA）和源 github.com/apache/spark/blob/…
因此，如果您确实需要按键分组，那么鉴于当前的实现 groupByKey 是您可以获得的最佳选择。还有其他方法（例如外部排序），但这些方法不会产生相同的结构。

【解决方案2】：

sc.parallelize(Array(("red", "zero"), ("yellow", "one"), ("red", "two")))
.map(t => (t._1,List(t._2)))
.reduceByKey(_:::_)
.collect()

Array[(String, List[String])] = Array((red,List(zero, two)), (yellow,List(one)))

【讨论】：