### 使用 Python 实现 MapReduce 来合并小文件 在处理大量小型文件时，使用 Hadoop 和 MapReduce 可以有效地管理和优化这些文件的处理过程[^1]。为了展示如何利用 Python 实现这一目标，下面提供了一个具体的例子。 #### 示例场景描述 假设有一个包含多个文本文件的小型文件集合，每个文件都存储了一定数量的单词列表。目的是将所有这些文件中的内容读取出来，并统计各个单词在整个数据集中出现的总次数。 #### Mapper 阶段设计 Mapper 的职责是从输入源中提取键值对，在这个案例里就是从单个文件中逐行读入并解析成 (word, 1) 形式的元组作为输出： ```python #!/usr/bin/env python3 import sys def main(): for line in sys.stdin: words = line.strip().split() for word in words: print(f"{word}\t1") if __name__ == "__main__": main() ``` 这段代码会遍历标准输入流里的每一行文字，去除首尾空白字符后拆分成词语数组；接着针对每一个词项打印出对应的计数值 "1"[^2]。 #### Reducer 阶段设计 Reducer 接收来自 Mappers 发送过来相同 key 下的所有 value 列表，计算其累加和再写出最终的结果记录： ```python #!/usr/bin/env python3 from collections import defaultdict import sys current_word = None word_count = 0 word_to_count = defaultdict(int) for line in sys.stdin: try: word, count_str = line.split('\t', 1) count = int(count_str) if current_word is not None and current_word != word: print(f'{current_word}\t{word_count}') word_count = 0 current_word = word word_count += count except ValueError as e: continue if current_word is not None: print(f'{current_word}\t{word_count}') ``` 上述脚本实现了 Reduce 功能：它接收由 map 函数产生的中间结果——即每遇到一个新的关键词就会更新累计数直到遇见下一个不同的关键字为止；最后还要记得把最后一个分组的信息也写出去[^3]。 #### 执行流程说明 当运行此类基于 MapReduce 架构的任务时，通常需要先准备好待处理的数据集并将它们上传至分布式文件系统如 HDFS 中去。之后启动作业提交工具指定 mapper 脚本路径、reducer 脚本路径以及其他必要的参数配置信息即可触发整个批处理工作流[^4]。