如何在PySpark代码中拆分年份，统计单词年度文章出现次数？

需要计算每个单词在每一年的文章中出现的次数，但拆分年份与单词时出现问题，日期会和第一个单词拼接在一起，错误结果如下：

['20191124,woman', 'stabbed', 'adelaide', 'shopping', 'centre', '20191204,economy', 'continue', 'teetering', 'edge', 'recession', '20200401,coronanomics', 'learnt', 'coronavirus', 'economy', '20200401,coronavirus', 'home', 'test', 'kits', 'selling', 'chinese', 'community', '20201015,coronavirus', 'pacific', 'economy', 'foriegn', 'aid', 'china', '20201016,china', 'builds', 'pig', 'apartment', 'blocks', 'guard', 'swine', 'flu', '20211216,economy', 'starts', 'bounce', 'unemployment', '20211224,online', 'shopping', 'rise', 'due', 'coronavirus', '20211229,china', 'close', 'encounters', 'elon', 'musks']

输入数据格式 #

20191124,woman stabbed adelaide shopping centre
20191204,economy continue teetering edge recession
20200401,coronanomics learnt coronavirus economy
20200401,coronavirus home test kits selling chinese community
20201015,coronavirus pacific economy foriegn aid china
20201016,china builds pig apartment blocks guard swine flu
20211216,economy starts bounce unemployment
20211224,online shopping rise due coronavirus
20211229,china close encounters elon musks

期望输出格式示例 #

adelaide    2019:1
aid    2020:1
apartment    2020:1
blocks    2020:1
...

当前编写的PySpark代码 #

from pyspark import SparkContext, SparkConf

sc = SparkContext('local','frequencies')

text = sc.textFile('abcnews.txt')

word_year_pairs = text.flatMap(lambda line: [((word, line[:4]), 1) for word in line.split()])

word_counts = word_counter.reduceByKey(lambda a,b: a+b)

result = word_counts.map(lambda pair: (pair[0][0] + '\t' + pair[0][1] + ":" + str(pair[1])))

final = result.sortByKey()

final.saveAsTextFile('rdd.py')

关键问题分析 #

原代码核心错误有两点：

1. 直接用line.split()拆分整行，导致逗号前的日期和第一个单词被合并成一个字符串（如20191124,woman），无法正确分离单词。
2. 虽然后续提取了年份line[:4]，但错误绑定给了包含日期+单词的无效"单词"，导致统计逻辑完全错误。

修正后的代码 #

from pyspark import SparkContext, SparkConf

sc = SparkContext('local','frequencies')

text = sc.textFile('abcnews.txt')

# 拆分每行的日期和正文，提取年份后将每个单词与年份配对
word_year_pairs = text.flatMap(lambda line: 
    [((word, date_part[:4]), 1) 
     for date_part, text_part in [line.split(',', 1)] 
     for word in text_part.split()]
)

# 按(单词,年份)键统计出现次数
word_counts = word_year_pairs.reduceByKey(lambda a, b: a + b)

# 转换为期望的输出格式
result = word_counts.map(lambda pair: f"{pair[0][0]}\t{pair[0][1]}:{pair[1]}")

# 按单词排序后保存结果
final = result.sortByKey()
final.saveAsTextFile('word_year_counts')

关键改动说明 #

• 拆分逻辑优化：用line.split(',', 1)将每行拆分为日期部分和正文部分，确保日期与第一个单词彻底分离。
• 正确绑定年份：从日期字符串提取前4位作为年份，再将正文拆分后的每个单词与该年份配对，生成有效的(单词,年份)统计键。
• 修正变量错误：原代码中word_counter是笔误，改为正确的word_year_pairs。
• 输出格式调整：用f-string简化格式拼接，确保单词与年份统计结果用制表符分隔，符合期望格式。

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