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

Question

Apache Spark pyspark.RDDAPI文档提到groupByKey()效率低下.相反,它是推荐使用reduceByKey(),aggregateByKey(),combineByKey(),或foldByKey()代替.这将导致在shuffle之前在worker中进行一些聚合,从而减少跨工作人员的数据混乱.

给定以下数据集和groupByKey()表达式,什么是等效且有效的实现(减少的跨工作者数据混洗),它不使用groupByKey(),但提供相同的结果？

dataset = [("a", 7), ("b", 3), ("a", 8)]
rdd = (sc.parallelize(dataset)
       .groupByKey())
print sorted(rdd.mapValues(list).collect())

输出:

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

Answer 1

据我所知,在这种特殊情况下,通过使用aggregateByKey或类似的功能,没有任何东西可以获得*.由于您正在构建一个列表,因此没有"真正的"减少,并且必须改组的数据量或多或少相同.

要真正观察某些性能增益,您需要实际上减少传输数据量的转换,例如计数,计算摘要统计信息,查找唯一元素.

关于使用不同的利益reduceByKey(),combineByKey()或者foldByKey()有一个重要的概念上的差异是比较容易看到当你考虑到Scala的API singatures.

既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].

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.Peters,T.PEP 20 - Python的禅宗.(2004年).在https://www.python.org/dev/peps/pep-0020/

*在实践中,这里实际上有很多东西要放松,特别是在使用PySpark时.Python的实现groupByKey比按键的天真组合更加优化.您可以查看由我和@eliasah创建的Be Smart关于groupByKey的其他讨论.