对增量数据做Map-Reduce¶
Map-reduce操作可以处理复杂的聚合任务。通过 mapReduce 命令和 mongo 命令行下封装好的 db.collection.mapReduce() 方法,都可以执行map-reduce操作。
如果数据集的记录数是不断增长的,你可能希望只对增量数据做map-reduce,而不是在整个数据集上面执行 map-reduce 。
对增量数据做map-reduce:
在当前的集合上执行 map-reduce ,并把结果输出到其他的集合。
当你有增量数据需要处理时,运行另外的 map-reduce:
在 query 参数中指定只查询新的增量文档:
在 out 参数中指定 reduce 函数来合并结果并输出到指定的集合中。
例如,你计划每天晚上在集合 sessions 上执行一次 map-reduce 操作。
创建数据¶
The sessions collection contains documents that log users’ sessions each day, for example:
db.sessions.save( { userid: "a", ts: ISODate('2011-11-03 14:17:00'), length: 95 } );
db.sessions.save( { userid: "b", ts: ISODate('2011-11-03 14:23:00'), length: 110 } );
db.sessions.save( { userid: "c", ts: ISODate('2011-11-03 15:02:00'), length: 120 } );
db.sessions.save( { userid: "d", ts: ISODate('2011-11-03 16:45:00'), length: 45 } );
db.sessions.save( { userid: "a", ts: ISODate('2011-11-04 11:05:00'), length: 105 } );
db.sessions.save( { userid: "b", ts: ISODate('2011-11-04 13:14:00'), length: 120 } );
db.sessions.save( { userid: "c", ts: ISODate('2011-11-04 17:00:00'), length: 130 } );
db.sessions.save( { userid: "d", ts: ISODate('2011-11-04 15:37:00'), length: 65 } );
在当前集合初始化Map-Reduce¶
首次运行map-reduce:
定义map函数,它映射用户的数据到新的文档中,新文档的字段有 userid , total_time , count , 和 avg_time :
var mapFunction = function() { var key = this.userid; var value = { userid: this.userid, total_time: this.length, count: 1, avg_time: 0 }; emit( key, value ); };
定义使用两个参数 key 和 values reduce函数,该函数会计算总时间和总个数。 key 对应的就是 userid , values 是一个数组,其中的元素就是上一步 mapFunction 函数输出的文档。
var reduceFunction = function(key, values) { var reducedObject = { userid: key, total_time: 0, count:0, avg_time:0 }; values.forEach( function(value) { reducedObject.total_time += value.total_time; reducedObject.count += value.count; } ); return reducedObject; };
定义使用两个参数 key 和 reducedValue 的完成函数。该方法在 reducedValue 文档中增加 average 字段,并返回修改后的文档。
var finalizeFunction = function (key, reducedValue) { if (reducedValue.count > 0) reducedValue.avg_time = reducedValue.total_time / reducedValue.count; return reducedValue; };
在 session 集合上执行map-reduce,包含 mapFunction , reduceFunction ,和 finalizeFunction 函数。将结果输出到 session_stat 集合中。如果 session_stat 集合已经存在,本次操作会先删除旧的内容:
db.sessions.mapReduce( mapFunction, reduceFunction, { out: "session_stat", finalize: finalizeFunction } )
对增量数据做Map-Reduce¶
以后,如果 sessions 有了新的数据记录,你可以运行增量数据的map-recude操作。例如,在 sessions 集合增加一些新的文档:
db.sessions.save( { userid: "a", ts: ISODate('2011-11-05 14:17:00'), length: 100 } );
db.sessions.save( { userid: "b", ts: ISODate('2011-11-05 14:23:00'), length: 115 } );
db.sessions.save( { userid: "c", ts: ISODate('2011-11-05 15:02:00'), length: 125 } );
db.sessions.save( { userid: "d", ts: ISODate('2011-11-05 16:45:00'), length: 55 } );
在每天快结束的时候,使用 query 选择新插入的文档,对这些文档执行 recude 后,再输出到 session_stat 集合中。
db.sessions.mapReduce( mapFunction,
reduceFunction,
{
query: { ts: { $gt: ISODate('2011-11-05 00:00:00') } },
out: { reduce: "session_stat" },
finalize: finalizeFunction
}
);