Python MapReduce入门求助：按key统计数量并计算value平均值

一、Python基础实现（本地运行，适合新手入门） #

先把输入数据整理成可处理的格式，用字典统计每个key的出现次数和value总和，最后计算平均值并输出：

# 输入数据
input_data = [
    "1,5",
    "1,5",
    "2,7",
    "2,8",
    "1,10",
    "2,10",
    "3,3",
    "1,20"
]

# 初始化统计字典：key为目标键，值为[出现次数, value总和]
stats = {}

for item in input_data:
    key, value = item.split(',')
    key = int(key)
    value = int(value)
    
    if key not in stats:
        stats[key] = [0, 0]
    stats[key][0] += 1  # 计数加1
    stats[key][1] += value  # 累加value

# 计算平均值并按key排序输出
for key in sorted(stats.keys()):
    count = stats[key][0]
    avg = stats[key][1] / count
    # 处理小数显示：整数直接输出，小数保留一位
    avg_display = round(avg, 1) if avg % 1 != 0 else int(avg)
    print(f"{key}, {count}, {avg_display}")

运行后输出和期望一致：

1, 4, 10
2, 3, 8.3
3, 1, 3

二、MapReduce实现（基于Hadoop Streaming） #

如果要在Hadoop集群上执行，可通过Python编写mapper和reducer脚本，用Hadoop Streaming提交任务：

1. Mapper脚本（mapper.py） #

负责拆分每行数据，输出key\tvalue格式供后续聚合：

#!/usr/bin/env python3
import sys

for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    key, value = line.split(',')
    print(f"{key}\t{value}")

2. Reducer脚本（reducer.py） #

负责聚合相同key的value，计算次数和平均值：

#!/usr/bin/env python3
import sys

current_key = None
current_count = 0
current_sum = 0

for line in sys.stdin:
    line = line.strip()
    key, value = line.split('\t')
    value = int(value)
    
    if current_key == key:
        current_count += 1
        current_sum += value
    else:
        if current_key is not None:
            # 输出上一个key的统计结果
            avg = current_sum / current_count
            avg_display = round(avg, 1) if avg % 1 != 0 else int(avg)
            print(f"{current_key}, {current_count}, {avg_display}")
        current_key = key
        current_count = 1
        current_sum = value

# 输出最后一个key的结果
if current_key is not None:
    avg = current_sum / current_count
    avg_display = round(avg, 1) if avg % 1 != 0 else int(avg)
    print(f"{current_key}, {current_count}, {avg_display}")

3. 运行步骤 #

1. 给脚本添加执行权限：

chmod +x mapper.py reducer.py

2. 将输入数据保存为input.txt，上传到HDFS：

hdfs dfs -put input.txt /user/yourname/input/

3. 提交Hadoop Streaming任务：

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \
    -files mapper.py,reducer.py \
    -mapper mapper.py \
    -reducer reducer.py \
    -input /user/yourname/input/input.txt \
    -output /user/yourname/output/

4. 查看输出结果：

hdfs dfs -cat /user/yourname/output/part-00000

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