Pyspark在嵌套元组上reduceByKey

我的问题与PySpark reduceByKey on multiple values类似，但有一点关键的区别。我是PySpark的新手，所以我肯定错过了一些明显的东西。Pyspark在嵌套元组上reduceByKey

我有以下结构的RDD：

(K0, ((k01,v01), (k02,v02), ...)) 
.... 
(Kn, ((kn1,vn1), (kn2,vn2), ...))

我想作为一个输出是一样的东西

(K0, v01+v02+...) 
... 
(Kn, vn1+vn2+...)

这似乎像是一个完美的情况下使用reduceByKey，起初我还以为类似的东西

rdd.reduceByKey(lambda x,y: x[1]+y[1])

这给了我完全RDD我开始用。我想我的索引有问题，因为有嵌套元组，但我尝试了所有可能的索引组合，我一直在想，并且一直给我回初始的RDD。

是否有理由不应该使用嵌套元组或者我做错了什么？

来源

2016-08-20 Violetta Zoffoli

这里根本不应该使用reduceByKey。它需要一个具有签名的关联和交换函数。 (T, T) => T。很显然，当您将List[Tuple[U, T]]作为输入并且您期望T作为输出时，它不适用。

由于不清楚是否键或唯一或不让我们考虑一般的例子，当我们必须在本地和全球聚合。让我们假设v01，v02，... vm是简单的数学运算：

from functools import reduce 
from operator import add 

def agg_(xs): 
    # For numeric values sum would be more idiomatic 
    # but lets make it more generic 
    return reduce(add, (x[1] for x in xs), zero_value) 

zero_value = 0 
merge_op = add 
def seq_op(acc, xs): 
    return acc + agg_(xs) 

rdd = sc.parallelize([ 
    ("K0", (("k01", 3), ("k02", 2))), 
    ("K0", (("k03", 5), ("k04", 6))), 
    ("K1", (("k11", 0), ("k12", -1)))]) 

rdd.aggregateByKey(0, seq_op, merge_op).take(2) 
## [('K0', 16), ('K1', -1)]

如果项已独特的简单mapValues就足够了：

from itertools import chain 

unique_keys = rdd.groupByKey().mapValues(lambda x: tuple(chain(*x))) 
unique_keys.mapValues(agg_).take(2) 
## [('K0', 16), ('K1', -1)]

来源

2016-08-20 16:03:52 zero323

那现在我很清楚。是的，键是唯一的，所以mapValues方法就是我所需要的。非常感谢你。 –

Pyspark在嵌套元组上reduceByKey

回答

相关问题