Pyspark - RDD 提取值进行聚合答案

【问题标题】：Pyspark - RDD extract values to aggregatePyspark - RDD 提取值进行聚合
【发布时间】：2021-02-17 15:25:01
【问题描述】：

使用 Pyspark，我正在尝试使用 RDD 根据该 RDD 的内容进行聚合。

我的 RDD 目前看起来像（显然有更多数据）：

[([u'User1', u'2'], 1), ([u'User2', u'2'], 1), ([u'User1', u'3'], 1)]

我想把它聚合成格式：

User1  5
User2  2

我正在努力与 RDD 进行交互，特别是 RDD 中的列表以获取这些数据。我还希望将其保留为 RDD，而不是将其转换为数据框。

谁能告诉我怎么做？

【问题讨论】：

标签： apache-spark pyspark rdd

【解决方案1】：

另一种解决方案，与@mck 非常相似，但更具可读性，是使用运算符add 而不是另一个 lambda 函数：

from operator import add 

rdd = sc.parallelize([("user1", "2"), ("user2", "2"), ("user1", "3")])
rdd = rdd.map(lambda x: (x[0], int(x[1])))
rdd = rdd.reduceByKey(add)

"""
>>> rdd.collect()
>>> Out[54]: [('user2', 2), ('user1', 5)]
"""

【讨论】：

【解决方案2】：

您可以将 RDD map 转换为 (user, value) 格式，然后执行 reduceByKey 以按用户分组并对值求和。

result = rdd.map(lambda x: (x[0][0], int(x[0][1]))).reduceByKey(lambda x, y: x + y)

result.collect()
# [('User2', 2), ('User1', 5)]

【讨论】：