传统的 MapReduce 如 Hadoop, 是以任务的形式进行的 — 获取一批数据, 提交给系统, 然后获取结果. 但是, 有一些统计的需求是即时的, 统计任务需要持续的运行, 一旦数据生成, 便立即发给统计任务处理, 生成的结果”推”给接收者.
以一个网站用户在线时长统计的需求为例子, 那么系统就有这几个部分:
数据接收
接收 Web Server(如 Apache/Nginx) 的 log, 例如使用 syslog.
Mapper(格式转换)
依次输入以行为单位的原始的 Apache log, 输出一条或者多条结构化的数据. 这个输出将出 Reducer 进行下一步处理.
Reducer(统计器)
不同的精度用不同的统计器, 因为统计结果必须在要求的精度时间内进行输出. 例如当精度要求是小时, 用户连续在线1个小时, 并且横跨在2个自然小时上, 那么, 统计结果应该是2条. 如果精度要求是天, 那么类似, 跨越自然天的数据应该被分割.
当 Reducer 的精度时间到达之后(如一个小时过完), Reducer 应该复位.
传统 Reducer 的输入是来自 Mapper, 但 Reducer 的输入来源应该包括其它的 Reducer. 例如, 按小时统计的 Reducer 的输出可以作为按天统计的 Reducer 的输入.
结果分发器
结果会以不同的形式发送出去, 如写成文件, 发邮件, 推送到其它系统…
结果的结构
有一种简单的数据库存储结构(先不考虑分表分库), 表的结构为:
time, timespan, key, val
UNIQUE(time, key)用户在线时长的数据这样存:
2012-05-25 09:12:20, 小时, ip1, 100s // ip1在线了100s, 从09:12:20开始
2012-05-25 12:24:10, 小时, ip1, 200s // ip1在线了100s, 从09:12:20开始
2012-05-25 09:12:20, 小时, ip1, 300s // ip1 2012-05-25 在线了300s, 但不是连续在线时间系统
根据上面的思想, 可以设计出一个即时流式数据的 MapReduce 系统, 也可以做一个代码框架. 但系统和框架的区别是, 系统包含了运行环境.
上面不同部分之间的通信会形成一种广义上的”队列”, 所以需要进行队列管理.