MongoDB增量mapReduce，仅选择新文档，在上次mapReduce之后添加

Question

假设我有一个包含如下文档的集合(只是简化的示例，但它应该显示方案):

> db.data.find()
{ "_id" : ObjectId("4e9c1f27aa3dd60ee98282cf"), "type" : "A", "value" : 11 }
{ "_id" : ObjectId("4e9c1f33aa3dd60ee98282d0"), "type" : "A", "value" : 58 }
{ "_id" : ObjectId("4e9c1f40aa3dd60ee98282d1"), "type" : "B", "value" : 37 }
{ "_id" : ObjectId("4e9c1f50aa3dd60ee98282d2"), "type" : "B", "value" : 1 }
{ "_id" : ObjectId("4e9c1f56aa3dd60ee98282d3"), "type" : "A", "value" : 85 }
{ "_id" : ObjectId("4e9c1f5daa3dd60ee98282d4"), "type" : "B", "value" : 12 }

现在我需要收集有关该集合的一些统计信息。例如:

db.data.mapReduce(function(){
        emit(this.type,this.value);
     },function(key,values){
        var total = 0;
        for(i in values) {total+=values[i]};
        return total;
     },
{out:'stat'})

将在“stat”集合中收集总计。

> db.stat.find()
{ "_id" : "A", "value" : 154 }
{ "_id" : "B", "value" : 50 }

此时一切都很完美，但我已经坚持下一步:

1. “数据”集合不断更新新数据(旧文档保持不变，仅插入，不更新)
2. 我想定期更新“stat”集合，但不想每次都查询整个“data”集合，因此我选择运行增量mapReduce
3. 在“data”集合中的每次插入时更新“stat”集合并且不使用mapReduce似乎不错，但实际情况比这个示例更复杂，我只想按需获取统计信息。
4. 为此，我应该只能查询在上次 mapReduce 之后添加的文档
5. 据我所知，我不能依赖 ObjectId 属性，只需存储最后一个，然后选择存储了 ObjectId > 的每个文档，因为 ObjectId 与 SQL 数据库中的自动增量 ID 不相等(例如，不同的分片将产生不同的 ObjectId)。
6. 我可以更改 ObjectId 生成器，但不确定如何在分片环境中做得更好

所以问题是:

有没有办法只选择在最后一个mapReduce之后添加的文档来运行增量mapReduce，或者可能有另一种策略来更新不断增长的集合的统计数据？

Answer 1

您可以缓存时间并将其用作下一次增量映射缩减的屏障。

我们正在工作中对此进行测试，它似乎有效。如果我错了，请纠正我，但是当跨分片进行插入时，您无法安全地执行映射缩减。版本变得不一致，您的 Map-Reduce 操作将失败。 (如果您找到解决方案，请告诉我!:)

我们使用批量插入，每 5 分钟一次。一旦完成所有批量插入，我们就可以像这样运行map-reduce(在Python中):

m = Code(<map function>)
r = Code(<reduce function>)

# pseudo code
end = last_time + 5 minutes

# Use time and optionally any other keys you need here
q = bson.SON([("date" : {"$gte" : last_time, "$lt" : end})])

collection.map_reduce(m, r, out=out={"reduce": <output_collection>}, query=q)

请注意，我们使用了reduce而不是merge，因为我们不想覆盖之前的内容；我们希望使用相同的reduce函数将旧结果和新结果结合起来。