当前位置：首页 > news >正文

使用 Hadoop Streaming 实现批处理分组

news 2026/8/26 18:05:06

在大数据处理中，我们经常需要对结构化数据（如 JSON 格式日志）按照某个字段进行分组，并对每个分组执行批量处理逻辑——例如聚合、写入数据库、调用外部服务等。Hadoop 的 MapReduce 模型天然支持按键分组，而Hadoop Streaming接口则允许我们使用 Python 等脚本语言灵活实现复杂的业务逻辑。

本文将介绍如何利用 Hadoop Streaming 和 Python 脚本，对每行包含 JSON 对象的大规模数据集，按指定字段进行分组，并以批处理方式处理每个分组，从而提升 I/O 效率或满足下游系统对批量操作的要求。

问题背景

假设输入数据是每行一个 JSON 对象的日志文件，例如：

{"user_id":"1001","action":"click","timestamp":1710000000}{"user_id":"1002","action":"view","timestamp":1710000005}{"user_id":"1001","action":"purchase","timestamp":1710000010}{"user_id":"1003","action":"click","timestamp":1710000020}

我们的目标是：

按user_id分组所有记录；
不逐条处理，而是将同一user_id的多条记录累积成批次；
当批次达到一定大小（如 1000 个分组）时，批量刷新（flush）到外部系统（如数据库、API 等）。

这种模式特别适用于：

减少数据库连接开销；
满足外部 API 的批量调用限制；
提高写入吞吐量。

基本原理

Hadoop MapReduce 的核心机制为该需求提供了天然支持：

Map 阶段：从每行 JSON 中提取目标字段（如user_id）作为 key，整行 JSON 作为 value。
Shuffle & Sort：Hadoop 自动将相同 key 的所有记录发送到同一个 Reducer，并按键排序。
Reduce 阶段：利用itertools.groupby对已排序的输入按键分组；再通过缓冲机制实现批量化处理。

关键点在于：Reducer 的输入已经按 key 排好序且分组完毕，我们只需控制“何时批量提交”。

实现步骤

1. 编写 Mapper 脚本

Mapper 负责解析 JSON 并输出(group_key, json_line)对：

importsysimportjsondefmain():forlineinsys.stdin:line=line.strip()ifnotline:continueobj=json.loads(line)key=obj.get("user_id")ifkeyisnotNone:print(f"{key}\t{line}")if__name__=='__main__':main()

注意：确保 key 不包含制表符或换行符，否则会破坏 Hadoop Streaming 的默认分隔规则。

2. 编写 Reducer 脚本（带批处理）

Reducer 使用groupby按 key 分组，并累积分组到缓冲区，达到阈值后批量处理：

importjsonimportsysfromitertoolsimportgroupby BATCH_SIZE=1000defflush_batch(buffer):forkey,recordsinbuffer:# 示例：打印分组信息print(f"Processing batch for key={key}, records={records}")# 在真实场景中，这里可能是：# - 批量插入数据库# - 发送 HTTP 请求# - 写入 Kafka 等defmain():buffer=[]forkey,groupingroupby(sys.stdin,key=lambdaline:line.split('\t',1)[0]):records=[]forlineingroup:parts=line.rstrip('\n').split('\t',1)iflen(parts)<2:continueobj=json.loads(parts[1])records.append(obj)buffer.append((key,records))iflen(buffer)>=BATCH_SIZE:flush_batch(buffer)buffer=[]ifbuffer:flush_batch(buffer)if__name__=='__main__':main()

说明：
groupby的 key 函数提取每行的 key（即\t前的部分）；
每个group是一个迭代器，包含该 key 下的所有行；
flush_batch是业务逻辑的入口，可根据需要替换。

3. 本地测试（可选）

在提交集群前，可在本地验证流程：

chmod+x mapper.py reducer.pycatinput.json|./mapper.py|sort|./reducer.py

注意：sort模拟了 Hadoop 的 shuffle 阶段，确保相同 key 相邻。

4. 提交到 Hadoop 集群

hadoop jar /path/to/hadoop-streaming.jar\-filesmapper.py,reducer.py\-mapper"python3 mapper.py"\-reducer"python3 reducer.py"\-input/user/input/json_logs\-output/user/output/grouped_batches