如何基于RDD通过ReduceByKey统计各榜单歌曲榜首总天数

你的思路不对，不能先按song单独分组——我们要统计的是同一首歌在不同榜单的独立榜首天数，核心是按「榜单类型+歌曲名称」的组合键来分组统计，具体步骤如下：

1. 转换数据结构 #

把每条过滤后的榜首记录，转换成以(chart_type, song)作为唯一标识的键，值固定为1（每条记录对应1天榜首）。
比如示例中的第一条记录：
('top200', '501', '1', '2021-03-26T00:00:00.000+02:00')
会被转换为：
(('top200', '501'), 1)

2. 按组合键聚合求和 #

使用reduceByKey对相同组合键的value进行求和，求和结果就是该歌曲在对应榜单的总榜首天数。

3. 格式化输出结果 #

把聚合后的键值对转换成你需要的chart_type, song, days in #1格式。

示例代码（PySpark） #

假设你已经有过滤好的RDD（仅包含排名为'1'的记录）：

# 转换为组合键+计数1的RDD
keyed_rdd = filtered_rdd.map(lambda record: ((record[0], record[1]), 1))
# 聚合求和
counted_rdd = keyed_rdd.reduceByKey(lambda a, b: a + b)
# 转换为目标输出格式
result_rdd = counted_rdd.map(lambda item: f"{item[0][0]}, {item[0][1]}, {item[1]}")

# 查看结果
for line in result_rdd.collect():
    print(line)

示例输出结果 #

top200, 501, 3
viral50, Gowtu, 3
top200, Lonely (with benny blanco), 3

内容的提问来源于stack exchange，提问作者user18972789