我有一个关于数据递增聚合的问题。
我有很多设备,存储在设备集合中,它们会发出(不规则的)测量值,这些值存储在记录集合中。每个记录都有一个时间戳(timestamp_utc),它不是存储数据的时间戳,而是测量值被测量的时间戳(因为设备发送了一堆测量值)。
我想要实现的是批量数据聚合(Map Reduce),对于每个不同的设备都有不同的比例尺度递增。例如,我想要一个包含平均每5分钟、每30分钟等的集合,以及每天的集合等等。我不想每次处理全部数据,而只想处理新的条目。
例如:记录集合包含:
{ _id : {device1, time : ISODate(2011-10-12T13:50:01Z)}, value : { meas1 : 2, meas2 : 4}},
{ _id : {device1, time : ISODate(2011-10-12T13:51:01Z)}, value : { meas1 : 1, meas2 : 6}},
{ _id : {device2, time : ISODate(2011-10-12T13:49:01Z)}, value : { meas1 : 3, meas2 : 7}},
{ _id : {device2, time : ISODate(2011-10-12T13:50:01Z)}, value : { meas1 : 4, meas2 : 8}},
{ _id : {device2, time : ISODate(2011-10-12T13:51:01Z)}, value : { meas1 : 5, meas2 : 9}},
在规模1上(每5分钟)进行数据聚合后,我将得到类似以下内容:
{ _id : {device1, time : ISODate(2011-10-12T13:50:00Z)}, value : { meas1 : 1.5, meas2 : 5}},
{ _id : {device2, time : ISODate(2011-10-12T13:45:00Z)}, value : { meas1 : 3, meas2 : 7}},
{ _id : {device2, time : ISODate(2011-10-12T13:50:00Z)}, value : { meas1 : 4.5, meas2 : 8.5}},
每次销售都会产生等等数据。 第一次生成这些数据通常不是什么大问题,这是一个非常普通的映射/缩减操作。
db.record.mapReduce(map, reduce, {finalize : finalize, out : { merge : db.recordscale1 }});
在进行增量式Map Reduce时会出现问题。实际上我想要做的是像这样:
db.record.mapReduce(map, reduce, {query : { "_id.time_utc" : { $gte : timeMin } }, finalize : finalize, out : { merge : db.recordscale1 }});
问题在于每个设备的timeMin都是不同的,它应该对应于最后一个插入到输出集合中(按照time_utc排序)的条目。
我尝试了很多技术但都没有成功(使用插入时间...)。 有人有想法如何处理吗? 我想避免为每个设备执行一次mapreduce,因为这将太低效。