MapReduce程序Groupby应用：Python处理大文件计算多维度商品均价

针对20万条记录的大文件，指定统计年份后按区域、产品类型两个维度分组计算组内商品平均价格，实现方案如下：

依赖说明 #

• 使用轻量级Python MapReduce库mrjob实现，无需依赖大数据集群，本地即可运行，也支持对接Hadoop处理更大规模数据
• 假设输入文件为逗号分隔的CSV格式，包含字段：年份、区域、产品类型、商品价格，可根据实际字段顺序调整解析逻辑

完整实现代码 #

from mrjob.job import MRJob
from mrjob.step import MRStep

class MRPriceAvg(MRJob):
    # 定义接收指定年份的入参
    def configure_args(self):
        super(MRPriceAvg, self).configure_args()
        self.add_passthru_arg('--target-year', type=int, required=True, help='指定需要统计的年份')

    def mapper(self, _, line):
        # 解析每行数据
        fields = line.strip().split(',')
        # 跳过表头（如果文件无表头可删除该行判断）
        if fields[0] == '年份':
            return
        year = int(fields[0])
        # 仅过滤目标年份的数据
        if year == self.options.target_year:
            area = fields[1]
            product_type = fields[2]
            price = float(fields[3])
            # 输出分组键为(区域, 产品类型)，值为(当前商品价格, 计数标记1)
            yield (area, product_type), (price, 1)

    def reducer(self, key, values):
        total_price = 0.0
        total_count = 0
        # 遍历同组所有记录，累加总价和总数量
        for price, cnt in values:
            total_price += price
            total_count += cnt
        # 计算平均价格，保留2位小数
        avg_price = round(total_price / total_count, 2)
        # 输出最终结果：区域、产品类型、对应平均价格
        yield key[0], (key[1], avg_price)

    def steps(self):
        return [
            MRStep(mapper=self.mapper, reducer=self.reducer)
        ]

if __name__ == '__main__':
    MRPriceAvg.run()

运行方法 #

1. 先安装依赖库：pip install mrjob
2. 执行统计命令，示例统计2023年的分组平均价格，结果输出到result.csv：
   python price_avg.py 你的输入文件路径.csv --target-year 2023 > result.csv

逻辑说明 #

• Map阶段：逐行读取数据过滤目标年份，按区域+产品类型生成分组键，输出每个商品的价格和计数标记
• Shuffle阶段（即MapReduce自带的Groupby特性）：框架自动将所有相同分组键的记录聚合到同一个Reducer处理，自动完成多维度分组逻辑，无需手动实现分组
• Reduce阶段：聚合同组所有商品的总价格和总数量，计算得到组内平均价格

注意事项 #

• 如果输入文件的字段顺序、分隔符和假设不一致，仅需修改mapper方法中的字段解析逻辑即可，整体计算逻辑无需调整
• 20万条记录规模很小，本地运行即可秒级出结果，不需要集群资源

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