在大数据处理场景中去除重复数据是一项常见且重要的任务。Hadoop 提供了强大的分布式计算能力而 Hadoop Streaming 接口允许我们使用任意编程语言如 Python编写 MapReduce 程序。本文将详细介绍如何通过 Hadoop Streaming 和 Python 脚本实现高效的数据去重。问题背景假设我们有一个大型文本文件每行代表一条记录例如日志条目、用户 ID、URL 等其中可能存在大量重复行。目标是生成一个仅包含唯一记录的输出文件。由于数据量可能达到 TB 级别单机处理效率低下因此借助 Hadoop 的分布式能力进行去重是合理的选择。基本原理Hadoop 的 MapReduce 模型天然适合去重任务Map 阶段将每行数据作为键key输出值value可以为空或任意占位符。Shuffle 阶段Hadoop 自动对相同 key 的记录进行分组并发送到同一个 Reducer。Reduce 阶段对于每个唯一的 key仅输出一次从而实现去重。由于 Hadoop 默认按键对中间结果进行排序和聚合相同的记录会被归并到一起这正是去重所需的核心机制。实现步骤1. 编写 Mapper 脚本Mapper 的作用是读取输入行并将其原样作为 key 输出。importsysforlineinsys.stdin:lineline.strip()ifline:print(f{line}\t)注意这里使用制表符\t分隔 key 和 value符合 Hadoop Streaming 的默认分隔规则。value 部分留空因为我们只关心 key。2. 编写 Reducer 脚本Reducer 接收已按键分组的输入对每个唯一 key 仅输出一次。importsys prev_keyNoneforlineinsys.stdin:lineline.strip()ifnotline:continuekeyline.split(\t)[0]ifkey!prev_key:print(key)prev_keykey该脚本通过比较当前 key 与前一个 key 是否相同确保每个唯一值只被打印一次。另一种简洁写法可使用itertools.groupby但需注意输入必须已排序Hadoop 保证这一点fromitertoolsimportgroupbyimportsysforkey,_ingroupby(sys.stdin,lambdax:x.split(\t)[0]):print(key)3. 本地测试可选在提交到集群前建议先在本地验证逻辑chmodx mapper.py reducer.pycatinput.txt|./mapper.py|sort|./reducer.py此命令模拟了 Map → Shufflesort→ Reduce 的完整流程。4. 提交到 Hadoop 集群使用 Hadoop Streaming JAR 包提交作业hadoop jar /path/to/hadoop-streaming.jar\-filesmapper.py,reducer.py\-mapperpython3 mapper.py\-reducerpython3 reducer.py\-input/user/input/data\-output/user/output/dedup注意事项-files参数将脚本分发到所有工作节点。根据集群环境可能需要指定python3的完整路径如/usr/bin/python3。确保所有节点已安装所需 Python 版本。扩展与优化按字段去重若只需对某列去重例如第二列可在 Mapper 中提取该字段作为 key# mapper_field.py 示例按第二列去重forlineinsys.stdin:fieldsline.strip().split(,)iflen(fields)2:print(f{fields[1]}\t{line.strip()})Reducer 则输出整行或指定内容。性能考虑对于高重复率数据可在 Map 端启用 Combiner通常与 Reducer 相同减少网络传输。输出结果会按键的字典序排列这是 Hadoop 的默认行为。示例效果输入文件内容apple banana apple cherry banana去重后输出apple banana cherry总结通过 Hadoop Streaming 与 Python 脚本我们可以快速构建一个分布式去重系统无需编写 Java 代码。该方法简单、灵活适用于各类基于行或字段的去重需求。在实际应用中可根据数据格式和业务逻辑调整 Mapper 和 Reducer 的逻辑以满足更复杂的去重策略如保留最新记录、按时间窗口去重等。