最近啃大数据的 MapReduce光看概念总觉得虚索性用 Python 纯手写了一遍完整流程。从最基础的单文件统计到大文件拆分、Map 局部计算、Reduce 汇总不用搭任何大数据框架就能把分治的核心逻辑摸得明明白白。这篇是我整理的实操笔记代码都拆成了小段每一步干啥都讲清楚入门练手很合适。一、先热个身小文件直接统计先从最简单的场景入手比如统计班级人数、统计各城市过车数量。逻辑都是统一的读数据 → 提取关键字段 → 字典计数。例子1统计每个班级的学生人数对应代码Demo01_student.py第一步先把文件内容读进来顺便去掉每行末尾的换行符with open(student.txt, moder, encodingutf-8) as f: # 逐行读取并去除首尾空白 students [line.strip() for line in f.readlines()]数据格式是姓名,年龄,班级我们只需要最后一列的班级信息。这里用 map 加 lambda 提取写起来比较简洁# 按逗号分割取每行最后一列的班级 clazzs list(map(lambda line: line.split(,)[-1], students))接下来就是核心的统计逻辑用字典存结果。key 是班级value 是人数遍历的时候判断不在字典里就初始化为 1已经存在就加 1clazz_num {} for clazz in clazzs: if clazz not in clazz_num: clazz_num[clazz] 1 else: clazz_num[clazz] 1 print(clazz_num)例子2统计每个城市的车流量对应代码Demo02_gateway_records.py和上面的逻辑完全一致只是把“班级”换成了“城市”。车辆数据里第三列是城市所以下标取 2 就行with open(gateway_records.txt, moder, encodingutf-8) as f: cars [line.strip() for line in f.readlines()] # 提取第三列城市 citys [line.split(,)[2] for line in cars] # 字典计数统计 city_num {} for city in citys: if city not in city_num: city_num[city] 1 else: city_num[city] 1 print(city_num)到这里都是基础操作几十兆的小文件随便跑。但如果是几个 G 甚至更大的文件一次性读进内存直接就崩了这就轮到 MapReduce 的分治思想出场。二、分治思路大文件拆成小块逐个处理MapReduce 的核心就是四个字分而治之。整套流程拆成三步Split 拆分把超大文件切成一堆小文件每个小文件都能单独放进内存Map 映射对每个小文件分别做统计生成局部结果Reduce 归约把所有局部结果汇总得到最终的全局结果打个比方数一仓库乒乓球一个人数太慢就分成好多小盒每个人数一盒这就是 Map最后把所有人的数加起来这就是 Reduce。三、Split 阶段大文件切分对应代码Demo03_split.py我们按行数切分比如每 10000 行存成一个小文件。跑之前记得先建好 split 文件夹不然写入会报错。先初始化变量分片编号、行数计数器并且打开第一个输出文件page 0 # 分片文件的序号 count 0 # 当前分片已经写了多少行 # 打开第一个分片文件准备写入 split_f open(fsplit/part-{page}, modew, encodingutf-8)然后循环读取大文件边读边往分片文件里写with open(gateway_records.txt, moder, encodingutf-8) as f: line f.readline() while line: # 读到文件末尾就退出循环 count 1 split_f.write(line) line f.readline()关键逻辑当行数达到 10000 时关闭当前文件序号 1打开新的分片文件计数器归零if count 10000: print(f已生成part-{page}) count 0 page 1 split_f.close() split_f open(fsplit/part-{page}, modew, encodingutf-8)跑完之后一个大文件就被切成了一堆part-x命名的小文件每个都是 10000 行。四、Map 阶段每个分片单独统计对应代码Demo04_map.py现在 split 文件夹里有一堆小文件了我们遍历每个文件各自统计城市数量把结果存到 map 文件夹里。同样提前建好 map 文件夹。先拿到 split 目录下所有的文件名import os base_dir split\\ files os.listdir(base_dir)循环处理每一个文件里面的统计逻辑和最开始的小文件统计一模一样for file in files: file_path base_dir file with open(file_path, moder, encodingutf-8) as f: lines [line.strip() for line in f.readlines()] # 提取城市列 citys [line.split(,)[2] for line in lines] # 当前分片的局部统计 city_num {} for city in citys: if city not in city_num: city_num[city] 1 else: city_num[city] 1统计完之后把这个分片的结果写入新文件格式是城市,数量# 保存当前分片的统计结果 with open(fmap\\{file}, modew, encodingutf-8) as w_f: for city, num in city_num.items(): w_f.write(f{city},{num}\n) w_f.flush() print(f{file_path}处理完成)这一步做完每个小文件都有了自己的统计结果。这就是 Map 阶段只负责自己这块数据的计算互不干扰。五、Reduce 阶段汇总所有结果对应代码Demo05_reduce.py最后一步把 Map 阶段所有文件的结果合起来相同城市的数量累加。记得建好 reduce 文件夹。还是先遍历 map 目录下的所有结果文件准备一个总字典存最终结果import os base_dir map\\ files os.listdir(base_dir) city_num {} # 最终结果字典逐行读取每个文件把城市和数量拆出来累加到总字典里for file in files: file_path base_dir file with open(file_path, moder, encodingutf-8) as f: lines [line.strip() for line in f.readlines()] for line in lines: city line.split(,)[0] num int(line.split(,)[1]) if city not in city_num: city_num[city] num else: city_num[city] num最后把最终结果保存下来就大功告成了with open(reduce/part-0, modew, encodingutf-8) as w_f: for city, num in city_num.items(): w_f.write(f{city},{num}\n) w_f.flush()六、附模拟数据生成脚本对应代码gen_data.py如果想自己测试大文件效果可以用这个脚本生成卡口过车数据。字段包括车牌号、卡口编号、城市、车辆品牌、道路编号、车速、行驶方向。比如生成随机车牌号的函数def generate_plate_number(): provinces [京] letters string.ascii_uppercase numbers .join(random.choices(string.digits, k5)) return random.choice(provinces) random.choice(letters) numbers生成城市、品牌就更简单了给个列表随机选def generate_city(): cities [北京, 上海, 广州, 深圳, 杭州, 成都, 武汉, 南京, 重庆, 天津] return random.choice(cities)最后主函数里控制生成条数直接写入文件if __name__ __main__: file_path gateway_records2.txt num_records 10000000 # 生成1千万条自己测试可以调小 generate_and_save_records(file_path, num_records) print(f已生成 {num_records} 条数据并保存到 {file_path})最后整套跑下来就能很直观地感受到 MapReduce 的思想拆分大任务 → 并行处理小任务 → 合并所有结果。真实的 Hadoop 里还涉及排序、分区、多节点调度这些但核心逻辑和我们手写的这个流程是一致的。用 Python 跑一遍小 demo再去学框架就会好理解很多。