执行增量映射-减少_MonogDB 中文网

MongoDB 中文手册

参考 > 聚合 > 映射减少 > 执行增量映射-减少

在本页面

资料设定
当前集合的初始Map-Reduce
随后的增量映射减少
聚合替代

聚合管道作为替代

聚合流水线比map-reduce提供更好的性能和更一致的接口。

各种地图-减少操作可以使用被重写聚合管道运营商，诸如$group， $merge等下面的例子包括聚集管道的替代方案。

为了执行map-reduce操作，MongoDB提供了 mapReduce命令，并在mongo外壳中提供了db.collection.mapReduce()wrapper方法。

如果map-reduce数据集不断增长，则可能需要执行增量map-reduce而不是每次都对整个数据集执行map-reduce操作。

要执行增量映射减少：

在当前集合上运行map-reduce作业，然后将结果输出到单独的集合。
当您有更多数据要处理时，请使用以下命令运行后续的map-reduce作业：
- 该query参数指定仅与新文档匹配的条件。
- 该out参数指定reduce将新结果合并到现有输出集合中的操作。

请考虑以下示例，在该示例中，您usersessions对每天要在集合上运行的map-reduce操作进行调度。

数据设置¶

该usersessions集合包含每天记录用户会话的文档，例如：

复制

db.usersessions.insertMany([
   { userid: "a", start: ISODate('2020-03-03 14:17:00'), length: 95 },
   { userid: "b", start: ISODate('2020-03-03 14:23:00'), length: 110 },
   { userid: "c", start: ISODate('2020-03-03 15:02:00'), length: 120 },
   { userid: "d", start: ISODate('2020-03-03 16:45:00'), length: 45 },
   { userid: "a", start: ISODate('2020-03-04 11:05:00'), length: 105 },
   { userid: "b", start: ISODate('2020-03-04 13:14:00'), length: 120 },
   { userid: "c", start: ISODate('2020-03-04 17:00:00'), length: 130 },
   { userid: "d", start: ISODate('2020-03-04 15:37:00'), length: 65 }
])

当前集合的初始Map- Reduce¶

运行第一个map-reduce操作，如下所示：

定义映射的映射函数userid到包含字段的对象total_time，count和avg_time：

复制

var mapFunction = function() {
    var key = this.userid;
    var value = { total_time: this.length, count: 1, avg_time: 0 };

    emit( key, value );
};

用两个参数定义相应的reduce函数， key并values计算总时间和计数。将key对应于userid和values是数组，其元素对应于映射到各个对象 userid中mapFunction。

复制

var reduceFunction = function(key, values) {

   var reducedObject = { total_time: 0, count:0, avg_time:0 };

   values.forEach(function(value) {
      reducedObject.total_time += value.total_time;
      reducedObject.count += value.count;
   });

   return reducedObject;
};

有两个参数定义finalize函数key和 reducedValue。该函数修改reducedValue文档以添加另一个字段average并返回修改后的文档。

复制

var finalizeFunction = function(key, reducedValue) {

   if (reducedValue.count > 0)
      reducedValue.avg_time = reducedValue.total_time / reducedValue.count;

   return reducedValue;
};

在执行映射简化usersessions使用收集 mapFunction的reduceFunction，和 finalizeFunction功能。将结果输出到集合 session_stats。如果session_stats集合已经存在，则该操作将替换内容：
复制
```
db.usersessions.mapReduce(
   mapFunction,
   reduceFunction,
   {
     out: "session_stats",
     finalize: finalizeFunction
   }
)
```

查询session_stats集合以验证结果：

复制

db.session_stats.find().sort( { _id: 1 } )

该操作返回以下文档：

{ "_id" : "a", "value" : { "total_time" : 200, "count" : 2, "avg_time" : 100 } }
{ "_id" : "b", "value" : { "total_time" : 230, "count" : 2, "avg_time" : 115 } }
{ "_id" : "c", "value" : { "total_time" : 250, "count" : 2, "avg_time" : 125 } }
{ "_id" : "d", "value" : { "total_time" : 110, "count" : 2, "avg_time" : 55 } }

随后的增量Map- Reduce¶

以后，随着usersessions集合的增长，您可以运行其他map-reduce操作。例如，将新文档添加到 usersessions集合中：

复制

db.usersessions.insertMany([
   { userid: "a", ts: ISODate('2020-03-05 14:17:00'), length: 130 },
   { userid: "b", ts: ISODate('2020-03-05 14:23:00'), length: 40 },
   { userid: "c", ts: ISODate('2020-03-05 15:02:00'), length: 110 },
   { userid: "d", ts: ISODate('2020-03-05 16:45:00'), length: 100 }
])

最终，对usersessions集合执行增量map-reduce ，但是使用该query字段仅选择新文档。将结果输出到collection session_stats，但是reduce将内容与增量map-reduce的结果进行比较：

复制

db.usersessions.mapReduce(
   mapFunction,
   reduceFunction,
   {
     query: { ts: { $gte: ISODate('2020-03-05 00:00:00') } },
     out: { reduce: "session_stats" },
     finalize: finalizeFunction
   }
);

查询session_stats集合以验证结果：

复制

db.session_stats.find().sort( { _id: 1 } )

该操作返回以下文档：

{ "_id" : "a", "value" : { "total_time" : 330, "count" : 3, "avg_time" : 110 } }
{ "_id" : "b", "value" : { "total_time" : 270, "count" : 3, "avg_time" : 90 } }
{ "_id" : "c", "value" : { "total_time" : 360, "count" : 3, "avg_time" : 120 } }
{ "_id" : "d", "value" : { "total_time" : 210, "count" : 3, "avg_time" : 70 } }

聚集替代¶

前提条件：将集合设置为原始状态：

复制

db.usersessions.drop();

db.usersessions.insertMany([
   { userid: "a", start: ISODate('2020-03-03 14:17:00'), length: 95 },
   { userid: "b", start: ISODate('2020-03-03 14:23:00'), length: 110 },
   { userid: "c", start: ISODate('2020-03-03 15:02:00'), length: 120 },
   { userid: "d", start: ISODate('2020-03-03 16:45:00'), length: 45 },
   { userid: "a", start: ISODate('2020-03-04 11:05:00'), length: 105 },
   { userid: "b", start: ISODate('2020-03-04 13:14:00'), length: 120 },
   { userid: "c", start: ISODate('2020-03-04 17:00:00'), length: 130 },
   { userid: "d", start: ISODate('2020-03-04 15:37:00'), length: 65 }
])

使用可用的聚合管道运算符，您可以重写map-reduce示例，而无需定义自定义函数：

复制

db.usersessions.aggregate([
   { $group: { _id: "$userid", total_time: { $sum: "$length" }, count: { $sum: 1 }, avg_time: { $avg: "$length" } } },
   { $project: { value: { total_time: "$total_time", count: "$count", avg_time: "$avg_time" } } },
   { $merge: {
      into: "session_stats_agg",
      whenMatched: [ { $set: {
         "value.total_time": { $add: [ "$value.total_time", "$$new.value.total_time" ] },
         "value.count": { $add: [ "$value.count", "$$new.value.count" ] },
         "value.avg": { $divide: [ { $add: [ "$value.total_time", "$$new.value.total_time" ] },  { $add: [ "$value.count", "$$new.value.count" ] } ] }
      } } ],
      whenNotMatched: "insert"
   }}
])

和$group分组userid，得出：

在total_time使用$sum操作
在count使用$sum操作
在avg_time使用$avg操作

该操作返回以下文档：

{ "_id" : "c", "total_time" : 250, "count" : 2, "avg_time" : 125 }
{ "_id" : "d", "total_time" : 110, "count" : 2, "avg_time" : 55 }
{ "_id" : "a", "total_time" : 200, "count" : 2, "avg_time" : 100 }
{ "_id" : "b", "total_time" : 230, "count" : 2, "avg_time" : 115 }

该$project阶段调整输出文档的形状以反映map-reduce的输出，该输出具有两个字段_id和 value。如果不需要镜像_idand value结构，则该阶段是可选的。

{ "_id" : "a", "value" : { "total_time" : 200, "count" : 2, "avg_time" : 100 } }
{ "_id" : "d", "value" : { "total_time" : 110, "count" : 2, "avg_time" : 55 } }
{ "_id" : "b", "value" : { "total_time" : 230, "count" : 2, "avg_time" : 115 } }
{ "_id" : "c", "value" : { "total_time" : 250, "count" : 2, "avg_time" : 125 } }

该$merge阶段将结果输出到 session_stats_agg集合。如果现有文档_id与新结果相同，则该操作将应用指定的管道，以根据结果和现有文档计算total_time，count和avg_time。如果是相同的，现有的文档_id中session_stats_agg，操作插入文档。

查询session_stats_agg集合以验证结果：

复制

db.session_stats_agg.find().sort( { _id: 1 } )

该操作返回以下文档：

{ "_id" : "a", "value" : { "total_time" : 200, "count" : 2, "avg_time" : 100 } }
{ "_id" : "b", "value" : { "total_time" : 230, "count" : 2, "avg_time" : 115 } }
{ "_id" : "c", "value" : { "total_time" : 250, "count" : 2, "avg_time" : 125 } }
{ "_id" : "d", "value" : { "total_time" : 110, "count" : 2, "avg_time" : 55 } }

将新文档添加到usersessions集合中：

复制

db.usersessions.insertMany([
   { userid: "a", ts: ISODate('2020-03-05 14:17:00'), length: 130 },
   { userid: "b", ts: ISODate('2020-03-05 14:23:00'), length: 40 },
   { userid: "c", ts: ISODate('2020-03-05 15:02:00'), length: 110 },
   { userid: "d", ts: ISODate('2020-03-05 16:45:00'), length: 100 }
])

$match在管道的开头添加一个阶段以指定日期过滤器：

复制

db.usersessions.aggregate([
   { $match: { ts: { $gte: ISODate('2020-03-05 00:00:00') } } },
   { $group: { _id: "$userid", total_time: { $sum: "$length" }, count: { $sum: 1 }, avg_time: { $avg: "$length" } } },
   { $project: { value: { total_time: "$total_time", count: "$count", avg_time: "$avg_time" } } },
   { $merge: {
      into: "session_stats_agg",
      whenMatched: [ { $set: {
         "value.total_time": { $add: [ "$value.total_time", "$$new.value.total_time" ] },
         "value.count": { $add: [ "$value.count", "$$new.value.count" ] },
         "value.avg_time": { $divide: [ { $add: [ "$value.total_time", "$$new.value.total_time" ] },  { $add: [ "$value.count", "$$new.value.count" ] } ] }
      } } ],
      whenNotMatched: "insert"
   }}
])

查询session_stats_agg集合以验证结果：

复制

db.session_stats_agg.find().sort( { _id: 1 } )

该操作返回以下文档：

{ "_id" : "a", "value" : { "total_time" : 330, "count" : 3, "avg_time" : 110 } }
{ "_id" : "b", "value" : { "total_time" : 270, "count" : 3, "avg_time" : 90 } }
{ "_id" : "c", "value" : { "total_time" : 360, "count" : 3, "avg_time" : 120 } }
{ "_id" : "d", "value" : { "total_time" : 210, "count" : 3, "avg_time" : 70 } }

可选的。为了避免$match每次运行时都必须修改聚合管道的日期条件，可以在帮助函数中定义包装聚合：

复制

updateSessionStats = function(startDate) {
   db.usersessions.aggregate([
      { $match: { ts: { $gte: startDate } } },
      { $group: { _id: "$userid", total_time: { $sum: "$length" }, count: { $sum: 1 }, avg_time: { $avg: "$length" } } },
      { $project: { value: { total_time: "$total_time", count: "$count", avg_time: "$avg_time" } } },
      { $merge: {
         into: "session_stats_agg",
         whenMatched: [ { $set: {
            "value.total_time": { $add: [ "$value.total_time", "$$new.value.total_time" ] },
            "value.count": { $add: [ "$value.count", "$$new.value.count" ] },
            "value.avg_time": { $divide: [ { $add: [ "$value.total_time", "$$new.value.total_time" ] },  { $add: [ "$value.count", "$$new.value.count" ] } ] }
         } } ],
         whenNotMatched: "insert"
      }}
   ]);
};

然后，要运行，您只需将开始日期传递给该updateSessionStats()函数：

复制

updateSessionStats(ISODate('2020-03-05 00:00:00'))

也可以看看