我想使用PySpark按值分组,然后在每个组中找到最大值。我有以下代码,但现在我在如何提取最大值方面卡住了。
# some file contains tuples ('user', 'item', 'occurrences')
data_file = sc.textData('file:///some_file.txt')
# Create the triplet so I index stuff
data_file = data_file.map(lambda l: l.split()).map(lambda l: (l[0], l[1], float(l[2])))
# Group by the user i.e. r[0]
grouped = data_file.groupBy(lambda r: r[0])
# Here is where I am stuck
group_list = grouped.map(lambda x: (list(x[1]))) #?
返回类似于以下内容:
[[(u'u1', u's1', 20), (u'u1', u's2', 5)], [(u'u2', u's3', 5), (u'u2', u's2', 10)]]
我现在希望找到每个用户的最大“occurrence”。在执行最大值后,最终结果将得到一个类似于以下RDD的结果:
[[(u'u1', u's1', 20)], [(u'u2', u's2', 10)]]
只保留文件中每个用户的最大数据集。换句话说,我想将RDD的值更改为仅包含单个三元组,即每个用户的最大出现次数。
(lambda x1, x2: max(x1, x2, key=lambda x: x[-1]))
吗? - WoodChoppermax
只是标准的Pythonmax
。它接受元素并返回最大的元素。key
参数描述了应该如何比较元素(这里是通过最后一个项目)。 - zero323