Pyspark 使用键计算值的出现次数

Question

我有一个(key,value)以下形式的对列表：

x=[(('cat','dog),('a','b')),(('cat','dog'),('a','b')),(('mouse','rat'),('e','f'))]

我想计算每个值元组与键元组一起出现的次数。

期望的输出：

[(('cat','dog'),('a','b',2)),(('mouse','rat'),('e','f',1))]

一个可行的解决方案是：

xs=sc.parallelize(x)
xs=xs.groupByKey()
xs=xs.map(lambda (x,y):(x,Counter(y))

然而，对于大型数据集，此方法会填满磁盘空间（~600GB）。我试图使用以下方法实现类似的解决方案reduceByKey：

xs=xs.reduceByKey(Counter).collect()

但我收到以下错误：

TypeError: __init__() takes at most 2 arguments (3 given)

Answer 1

我通常是这样做的：

xs=sc.parallelize(x)
a = xs.map(lambda x: (x, 1)).reduceByKey(lambda a,b: a+b)

a.collect()产量：

[((('mouse', 'rat'), ('e', 'f')), 1), ((('cat', 'dog'), ('a', 'b')), 2)]

我假设您想要 (key1, key2) 对中的第二个键内的计数（此处为 1 和 2）。

要实现这一目标，请尝试以下操作：

a.map(lambda x: (x[0][0], x[0][1] + (x[1],))).collect()

最后一步基本上是重新映射它，以便您获得第一个密钥对（如('mouse','rat')），然后获取第二个密钥对（如('e','f')），然后将的tuple版本b[1]（即计数）添加到第二个密钥对。