即时流式数据 MapReduce

浏览数：27 / 时间：2015年06月08日

传统的 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, timespan, 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, 但不是连续在线时间