问Apache Spark:使用RDD.aggregateByKey()实现RDD.groupByKey()的等价物是什么？

EN

据我所知，在这种特殊情况下，使用aggregateByKey或类似的函数没有什么好处。由于您正在构建一个列表，因此没有“真正的”缩减，并且必须混洗的数据量大致相同。

要真正观察到一些性能增益，您需要进行转换，这些转换实际上减少了传输的数据量，例如计数、计算汇总统计信息、查找唯一元素。

关于使用reduceByKey()、combineByKey()或foldByKey()的区别和好处，有一个重要的概念上的区别，当您考虑Scala API特征时，这一点更容易看出。

reduceByKey和foldByKey都从RDD[(K, V)]映射到RDD[(K, V)]，而第二个提供了额外的零元素。

reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)] 
foldByKey(zeroValue: V)(func: (V, V) ⇒ V): RDD[(K, V)]

combineByKey (没有aggregateByKey，但它是同一类型的转换)从RDD[(K, V)]转换为RDD[(K, C)]

combineByKey[C](
   createCombiner: (V) ⇒ C,
   mergeValue: (C, V) ⇒ C,
   mergeCombiners: (C, C) ⇒ C): RDD[(K, C)] 

返回到您的示例，只有combineByKey (在PySpark aggregateByKey中)才是真正适用的，因为您正在从RDD[(String, Int)]转换到RDD[(String, List[Int])]。

虽然在Python这样的动态语言中，使用foldByKey或reduceByKey执行这样的操作实际上是可能的，但这会使代码的语义变得不清楚，并引用@tim-peters的话说：“应该有1个，最好只有一个--显而易见的方法”1。

aggregateByKey和combineByKey之间的区别与reduceByKey和foldByKey之间的区别基本相同，所以对于一个列表来说，这主要是一个品味问题：

def merge_value(acc, x):
    acc.append(x)
    return acc

def merge_combiners(acc1, acc2):
    acc1.extend(acc2)
    return acc1

rdd = (sc.parallelize([("a", 7), ("b", 3), ("a", 8)])
   .combineByKey(
       lambda x: [x],
       lambda u, v: u + [v],
       lambda u1,u2: u1+u2))

但在实践中，您应该更喜欢groupByKey。与上面提供的简单实现相比，PySpark实现要优化得多。

1. 1.Peters，T. PEP 20 -- Python的禅宗。(2004)。在https://www.python.org/dev/peps/pep-0020/

*在实践中，这里实际上有相当多的松散，特别是在使用PySpark时。groupByKey的Python实现比naive combine by key要优化得多。你可以查看由我和@eliasah创建的Be Smart About groupByKey来进行更多的讨论。

这里有一个使用aggregateByKey()的选项。我很想知道如何使用reduceByKey()、combineByKey()或foldByKey()来实现这一点，以及每种替代方案的成本/收益是什么。

rdd = (sc.parallelize([("a", 7), ("b", 3), ("a", 8)])
       .aggregateByKey(list(),
                       lambda u,v: u+[v],
                       lambda u1,u2: u1+u2))
print sorted(rdd.mapValues(list).collect())

输出：

[('a', [7, 8]), ('b', [3])]

下面是一个内存效率稍高的实现，尽管python新手的可读性较差，但会产生相同的输出：

rdd = (sc.parallelize([("a", 7), ("b", 3), ("a", 8)])
       .aggregateByKey(list(),
                       lambda u,v: itertools.chain(u,[v]),
                       lambda u1,u2: itertools.chain(u1,u2)))
print sorted(rdd.mapValues(list).collect())