我在 Python 中遇到关于 Apache Spark 的问题。我有这套
data = sc.parallelize([('a','u'), ('a', 'v'),
('b', 'w'), ('b', 'x'), ('b', 'x')] )
我想做的是按键计算元素的数量并创建一个包含元素的列表。如果我这样做
a = data.combineByKey(lambda value: (value, 1),
lambda x, value: (value, x[1] + 1),
lambda x, y: (x[0]+'/'+y[0], x[1] + y[1]))
我有这样的结果:
[('a', ('u/v', 2)), ('b', ('w/x/x', 3))]
我想要的是
[('a', (['u','v'], 2)), ('b', (['w','x','x'], 3))]
我该怎么做？
请您参考如下方法:
如果您想将所有值保存为列表，则根本没有理由使用 combineByKey。简单地 groupBy 更有效:
aggregated = data.groupByKey().mapValues(lambda vs: (list(vs), len(vs)))
aggregated.collect()
## [('a', (['u', 'v'], 2)), ('b', (['w', 'x', 'x'], 3))]
一种更有效的方法是保留计数而不是所有值:
aggregated_counts = (data
.map(lambda kv: (kv, 1))
.reduceByKey(add)
.map(lambda kv: (kv[0][0], (kv[0][1], kv[1])))
.groupByKey()
.mapValues(lambda xs: (list(xs), sum(x[1] for x in xs))))
aggregated_counts.collect()
## [('a', ([('v', 1), ('u', 1)], 2)), ('b', ([('w', 1), ('x', 2)], 3))]
或
from collections import Counter
def merge_value(acc, x):
acc.update(x)
return acc
def merge_combiners(acc1, acc2):
acc1.update(acc2)
return acc1
aggregated_counts_ = (data
.combineByKey(Counter, merge_value, merge_combiners)
.mapValues(lambda cnt: (cnt, sum(cnt.values()))))
aggregated_counts_.collect()
## [('a', (Counter({'u': 1, 'v': 1}), 2)), ('b', (Counter({'w': 1, 'x': 2}), 3))]
