在mongoDB中使用时间间隔增量MapReduce

Question

我得到了来自服务器的一些记录有10分钟的时间间隔(1小时,我会得到6个文件),我想要做的地图减少在接下来的几个小时,每1小时,我将不得不做下一组的地图减少6个文件与去年小时文件我将如何解决这个问题？帮助我在过去1个月里混淆了frm谢谢你Sushil Kr Singh

Answer 1

为了按小时汇总10分钟的日志文件,您可以将每个日志文件的时间戳向下舍入到map函数中的最近小时,并在reduce函数中按小时对结果进行分组.

这是一个小的虚拟示例,它从mongo shell中说明了这一点:

1. 创建100个日志文件,每个文件相隔10分钟并包含0-10之间的随机数,并将它们插入logs数据库中的集合中:

   for (var i = 0; i < 100; i++) { 
       d = new ISODate();
       d.setMinutes(d.getMinutes() + i*10);
       r = Math.floor(Math.random()*11)
       db.logs.insert({timestamp: d, number: r})
   }
   

   要检查logs集合的外观,请发送一个类似的查询db.logs.find().limit(3).pretty(),其结果是:

   {
       "_id" : ObjectId("50455a3570537f9433c1efb2"),
       "timestamp" : ISODate("2012-09-04T01:32:37.370Z"),
       "number" : 2
   }
   {
       "_id" : ObjectId("50455a3570537f9433c1efb3"),
       "timestamp" : ISODate("2012-09-04T01:42:37.370Z"),
       "number" : 3
   }
   {
       "_id" : ObjectId("50455a3570537f9433c1efb4"),
       "timestamp" : ISODate("2012-09-04T01:52:37.370Z"),
       "number" : 8
   }
   

2. 定义一个映射函数(在此示例中称为mapf),将时间戳舍入到最接近的小时(向下舍入),该小时用于发出键.emit值是该日志文件的编号.

   mapf = function () { 
       // round down to nearest hour
       d = this.timestamp;
       d.setMinutes(0);
       d.setSeconds(0);
       d.setMilliseconds(0);
       emit(d, this.number); 
   }
   

3. 定义一个reduce函数,它对所有发出的值(即数字)求和.

   reducef = function (key, values) {
       var sum = 0;
       for (var v in values) {
           sum += values[v];
       }
       return sum;
   }
   

4. 现在在logs集合上执行map/reduce.out此处的参数指定我们要将结果写入hourly_logs集合并将现有文档与新结果合并.这可确保稍后提交的日志文件(例如,在服务器发生故障或其他延迟之后)一旦出现在日志中,就会包含在结果中.

   db.logs.mapReduce(mapf, reducef, {out: { merge : "hourly_logs" }})
   

5. 最后,要查看结果,您可以查询简单的查找hourly_logs:

   db.hourly_logs.find()
   
   { "_id" : ISODate("2012-09-04T02:00:00Z"), "value" : 33 }
   { "_id" : ISODate("2012-09-04T03:00:00Z"), "value" : 31 }
   { "_id" : ISODate("2012-09-04T04:00:00Z"), "value" : 21 }
   { "_id" : ISODate("2012-09-04T05:00:00Z"), "value" : 40 }
   { "_id" : ISODate("2012-09-04T06:00:00Z"), "value" : 26 }
   { "_id" : ISODate("2012-09-04T07:00:00Z"), "value" : 26 }
   { "_id" : ISODate("2012-09-04T08:00:00Z"), "value" : 25 }
   { "_id" : ISODate("2012-09-04T09:00:00Z"), "value" : 46 }
   { "_id" : ISODate("2012-09-04T10:00:00Z"), "value" : 27 }
   { "_id" : ISODate("2012-09-04T11:00:00Z"), "value" : 42 }
   { "_id" : ISODate("2012-09-04T12:00:00Z"), "value" : 43 }
   { "_id" : ISODate("2012-09-04T13:00:00Z"), "value" : 35 }
   { "_id" : ISODate("2012-09-04T14:00:00Z"), "value" : 22 }
   { "_id" : ISODate("2012-09-04T15:00:00Z"), "value" : 34 }
   { "_id" : ISODate("2012-09-04T16:00:00Z"), "value" : 18 }
   { "_id" : ISODate("2012-09-04T01:00:00Z"), "value" : 13 }
   { "_id" : ISODate("2012-09-04T17:00:00Z"), "value" : 25 }
   { "_id" : ISODate("2012-09-04T18:00:00Z"), "value" : 7 }
   

结果是10分钟日志的每小时摘要,_id字段包含小时的开头,值字段包含随机数的总和.在您的情况下,您可能有不同的聚合运算符; 根据您的需要修改reduce功能.

正如Sammaye在评论中提到的那样,您可以使用cron作业条目自动执行map/reduce调用以每小时运行一次.

如果您不希望每次都处理完整的日志集合,则可以通过将文档限制为每小时时间窗口来运行增量更新,如下所示:

var q = { $and: [ {timestamp: {$gte: new Date(2012, 8, 4, 12, 0, 0) }},                                                  
                  {timestamp: {$lt:  new Date(2012, 8, 4, 13, 0, 0) }} ] }

db.logs.mapReduce(mapf, reducef, {query: q, out: { merge : "hourly_logs" }})

这将只包括12到13小时之间的日志文件.请注意,Date()对象中的月份值从0(8 = 9月)开始.由于该merge选项,在已处理的日志文件上运行m/r是安全的.