如何在PySpark中对RDD分组统计，获取学生考试及格次数？

输入数据 #

new_data = [{'name': 'Tom', 'subject': "maths", 'exam_score': 85},
            {'name': 'Tom', 'subject': "science", 'exam_score': 55},
            {'name': 'Tom', 'subject': "history", 'exam_score': 68},
            {'name': 'Ivy', 'subject': "maths", 'exam_score': 72},
            {'name': 'Ivy', 'subject': "science", 'exam_score': 67},
            {'name': 'Ivy', 'subject': "history", 'exam_score': 59},
            {'name': 'Ben', 'subject': "maths", 'exam_score': 56},
            {'name': 'Ben', 'subject': "science", 'exam_score': 51},
            {'name': 'Ben', 'subject': "history", 'exam_score': 63},
            {'name': 'Eve', 'subject': "maths", 'exam_score': 74},
            {'name': 'Eve', 'subject': "maths", 'exam_score': 87},
            {'name': 'Eve', 'subject': "maths", 'exam_score': 90}]

new_rdd = sc.parallelize(new_data)

需求说明 #

规定考试得分≥60分为及格，需返回一个Spark RDD，存储每个学生的姓名及其及格的考试次数（次数范围1-3），要求使用groupByKey()和map()方法实现。

实现代码 #

# 转换为(name, 及格标记)的键值对：及格记1，不及格记0
pass_marks_rdd = new_rdd.map(lambda item: (item['name'], 1 if item['exam_score'] >= 60 else 0))

# 按姓名分组后，对每组的及格标记求和得到及格次数
result_rdd = pass_marks_rdd.groupByKey().map(lambda pair: (pair[0], sum(pair[1])))

# 输出结果
print(result_rdd.collect())

代码解释 #

1. map转换：遍历每条考试记录，提取学生姓名，根据分数判断是否及格并生成对应标记，将数据转换为(姓名, 1/0)的键值对格式。
2. groupByKey分组：将同一学生的所有及格标记聚合到同一分组中。
3. map求和：对每个学生的及格标记列表求和，得到该学生的总及格次数，最终生成(姓名, 及格次数)的结果元组。

预期输出 #

# [('Tom', 2),
# ('Ivy', 2),
# ('Ben', 1),
# ('Eve', 3)]

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