/ /

Map-Reduce 示例

注意

作为 Map-Reduce 替代方案的聚合管道

聚合管道提供比映射化简操作更好的性能和可用性。

可以使用聚合管道阶段（例如 $group 和 $merge）重写 map-reduce 操作。

对于需要自定义功能的 map-reduce 操作，MongoDB 提供了 $accumulator 和 $function 聚合操作符。可以使用这些操作符在 JavaScript 中定义自定义聚合表达式。

在 mongosh 中，db.collection.mapReduce() 方法是 mapReduce 命令的封装器。以下示例使用 db.collection.mapReduce() 方法。

本节中的示例包括不带自定义聚合表达式的聚合管道替代方案。有关使用自定义表达式的替代方案，请参阅Map-Reduce 到聚合管道转换示例。

创建一个包含以下文档的样本集合 orders：

db.orders.insertMany([
   { _id: 1, cust_id: "Ant O. Knee", ord_date: new Date("2020-03-01"), price: 25, items: [ { sku: "oranges", qty: 5, price: 2.5 }, { sku: "apples", qty: 5, price: 2.5 } ], status: "A" },
   { _id: 2, cust_id: "Ant O. Knee", ord_date: new Date("2020-03-08"), price: 70, items: [ { sku: "oranges", qty: 8, price: 2.5 }, { sku: "chocolates", qty: 5, price: 10 } ], status: "A" },
   { _id: 3, cust_id: "Busby Bee", ord_date: new Date("2020-03-08"), price: 50, items: [ { sku: "oranges", qty: 10, price: 2.5 }, { sku: "pears", qty: 10, price: 2.5 } ], status: "A" },
   { _id: 4, cust_id: "Busby Bee", ord_date: new Date("2020-03-18"), price: 25, items: [ { sku: "oranges", qty: 10, price: 2.5 } ], status: "A" },
   { _id: 5, cust_id: "Busby Bee", ord_date: new Date("2020-03-19"), price: 50, items: [ { sku: "chocolates", qty: 5, price: 10 } ], status: "A"},
   { _id: 6, cust_id: "Cam Elot", ord_date: new Date("2020-03-19"), price: 35, items: [ { sku: "carrots", qty: 10, price: 1.0 }, { sku: "apples", qty: 10, price: 2.5 } ], status: "A" },
   { _id: 7, cust_id: "Cam Elot", ord_date: new Date("2020-03-20"), price: 25, items: [ { sku: "oranges", qty: 10, price: 2.5 } ], status: "A" },
   { _id: 8, cust_id: "Don Quis", ord_date: new Date("2020-03-20"), price: 75, items: [ { sku: "chocolates", qty: 5, price: 10 }, { sku: "apples", qty: 10, price: 2.5 } ], status: "A" },
   { _id: 9, cust_id: "Don Quis", ord_date: new Date("2020-03-20"), price: 55, items: [ { sku: "carrots", qty: 5, price: 1.0 }, { sku: "apples", qty: 10, price: 2.5 }, { sku: "oranges", qty: 10, price: 2.5 } ], status: "A" },
   { _id: 10, cust_id: "Don Quis", ord_date: new Date("2020-03-23"), price: 25, items: [ { sku: "oranges", qty: 10, price: 2.5 } ], status: "A" }
])

返回每位客户的总价

针对 orders 集合执行 map-reduce 操作会按 cust_id 进行分组，并为每个 cust_id 计算 price 之和：

定义 Map 函数来处理每个输入文档：
- 在函数中，this 指的是 Map-Reduce 操作正在处理的文档。
- 该函数将每个文档的 price 映射到 cust_id，并输出 cust_id 和 price。
```
var mapFunction1 = function() {
   emit(this.cust_id, this.price);
};
```
使用两个参数 keyCustId 和 valuesPrices 定义相应的 Reduce 函数：
- valuesPrices 是一个数组，其元素是由 Map 函数发出并按 keyCustId 分组的 price 值。
- 该函数将 valuesPrice 数组缩减为其元素之和。
```
var reduceFunction1 = function(keyCustId, valuesPrices) {
   return Array.sum(valuesPrices);
};
```
使用 mapFunction1 Map 函数和 reduceFunction1 Reduce 函数对 orders 集合中的所有文档执行 Map-Reduce：
```
db.orders.mapReduce(
   mapFunction1,
   reduceFunction1,
   { out: "map_reduce_example" }
)
```
此操作将结果输出到名为 map_reduce_example 的集合。如果 map_reduce_example 集合已存在，该操作将用此 Map-Reduce 操作的结果替换其内容。
查询 map_reduce_example 集合以验证结果：
```
db.map_reduce_example.find().sort( { _id: 1 } )
```
该操作会返回以下文档：
```
{ "_id" : "Ant O. Knee", "value" : 95 }
{ "_id" : "Busby Bee", "value" : 125 }
{ "_id" : "Cam Elot", "value" : 60 }
{ "_id" : "Don Quis", "value" : 155 }
```

聚合替代方案

利用可用的聚合管道操作符，您可以重写 Map-Reduce 操作，而无需定义自定义函数：

db.orders.aggregate([
   { $group: { _id: "$cust_id", value: { $sum: "$price" } } },
   { $out: "agg_alternative_1" }
])

$group 阶段按 cust_id 分组并计算 value 字段（另见 $sum）。value 字段包含每个 cust_id 的 price 总额。
此阶段将以下文档输出到下一阶段：
```
{ "_id" : "Don Quis", "value" : 155 }
{ "_id" : "Ant O. Knee", "value" : 95 }
{ "_id" : "Cam Elot", "value" : 60 }
{ "_id" : "Busby Bee", "value" : 125 }
```
然后，$out 将输出写入集合 agg_alternative_1。或者，你可以使用 $merge 代替 $out。
查询 agg_alternative_1 集合以验证结果：
```
db.agg_alternative_1.find().sort( { _id: 1 } )
```
该操作将返回以下文档：
```
{ "_id" : "Ant O. Knee", "value" : 95 }
{ "_id" : "Busby Bee", "value" : 125 }
{ "_id" : "Cam Elot", "value" : 60 }
{ "_id" : "Don Quis", "value" : 155 }
```

提示

有关使用自定义聚合表达式的替代方案，请参阅 Map-Reduce 到聚合管道转换示例。

通过每款商品的平均数量计算订单和总数量

在以下示例中，您将看到针对 ord_date 值大于或等于 2020-03-01 的所有文档对 orders 集合执行的 Map-Reduce 操作。

示例中的操作：

按 item.sku 字段分组，计算每个 sku 的订单数量和总订购量。
计算每个 sku 值的每个订单的平均数量，并将结果合并到输出集合中。

合并结果时，如果现有文档与新结果具有相同的键，则该操作将覆盖现有文档。如果没有具有相同键的现有文档，操作将插入该文档。

步骤示例：

定义 Map 函数来处理每个输入文档：

在函数中，this 指的是 Map-Reduce 操作正在处理的文档。
对于每款商品，该函数将 sku 与新对象 value 关联，该对象包含 1 的 count 和订单的商品 qty，并发出 sku（存储在 key 中）和 value。

 var mapFunction2 = function() {
    for (var idx = 0; idx < this.items.length; idx++) {
       var key = this.items[idx].sku;
       var value = { count: 1, qty: this.items[idx].qty };
       emit(key, value);
    }
};

使用两个参数 keySKU 和 countObjVals 定义相应的 Reduce 函数：

countObjVals 是一个数组，其元素是映射到分组 keySKU 值的对象，这些值由 Map 函数传递到 Reducer 函数。
该函数将 countObjVals 数组缩减为包含 count 和 qty 字段的单个对象 reducedValue。
在 reducedVal 中，count 字段包含各个数组元素中 count 字段的总和，而 qty 字段包含各个数组元素中 qty 字段的总和。

var reduceFunction2 = function(keySKU, countObjVals) {
   reducedVal = { count: 0, qty: 0 };
   for (var idx = 0; idx < countObjVals.length; idx++) {
       reducedVal.count += countObjVals[idx].count;
       reducedVal.qty += countObjVals[idx].qty;
   }
   return reducedVal;
};

使用两个参数 key 和 reducedVal 定义 finalize 函数。该函数会修改 reducedVal 对象以添加名为 avg 的计算字段并返回修改后的对象：
```
var finalizeFunction2 = function (key, reducedVal) {
  reducedVal.avg = reducedVal.qty/reducedVal.count;
  return reducedVal;
};
```
使用 mapFunction2、reduceFunction2 和 finalizeFunction2 函数对 orders 集合执行 Map-Reduce 操作：
```
db.orders.mapReduce(
   mapFunction2,
   reduceFunction2,
   {
     out: { merge: "map_reduce_example2" },
     query: { ord_date: { $gte: new Date("2020-03-01") } },
     finalize: finalizeFunction2
   }
 );
```
此操作使用 query 字段来仅选择 ord_date 大于或等于 new Date("2020-03-01") 的文档。然后它将结果输出到集合 map_reduce_example2。
如果 map_reduce_example2 集合已存在，该操作会将现有内容与此 Map-Reduce 操作的结果合并。即，如果现有文档的键与新结果相同，则操作会覆盖现有文档。如果没有具有相同键的现有文档，操作将插入该文档。

查询 map_reduce_example2 集合以验证结果：

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

该操作会返回以下文档：

{ "_id" : "apples", "value" : { "count" : 4, "qty" : 35, "avg" : 8.75 } }
{ "_id" : "carrots", "value" : { "count" : 2, "qty" : 15, "avg" : 7.5 } }
{ "_id" : "chocolates", "value" : { "count" : 3, "qty" : 15, "avg" : 5 } }
{ "_id" : "oranges", "value" : { "count" : 7, "qty" : 63, "avg" : 9 } }
{ "_id" : "pears", "value" : { "count" : 1, "qty" : 10, "avg" : 10 } }

聚合替代方案

利用可用的聚合管道操作符，您可以重写 Map-Reduce 操作，而无需定义自定义函数：

db.orders.aggregate( [
   { $match: { ord_date: { $gte: new Date("2020-03-01") } } },
   { $unwind: "$items" },
   { $group: { _id: "$items.sku", qty: { $sum: "$items.qty" }, orders_ids: { $addToSet: "$_id" } }  },
   { $project: { value: { count: { $size: "$orders_ids" }, qty: "$qty", avg: { $divide: [ "$qty", { $size: "$orders_ids" } ] } } } },
   { $merge: { into: "agg_alternative_3", on: "_id", whenMatched: "replace",  whenNotMatched: "insert" } }
] )

$match 阶段仅选择 ord_date 大于等于 new Date("2020-03-01") 的文档。

$unwind 阶段按 items 数组字段对文档进行分解，为每个数组元素输出一个文档。例如：

{ "_id" : 1, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-01T00:00:00Z"), "price" : 25, "items" : { "sku" : "oranges", "qty" : 5, "price" : 2.5 }, "status" : "A" }
{ "_id" : 1, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-01T00:00:00Z"), "price" : 25, "items" : { "sku" : "apples", "qty" : 5, "price" : 2.5 }, "status" : "A" }
{ "_id" : 2, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 70, "items" : { "sku" : "oranges", "qty" : 8, "price" : 2.5 }, "status" : "A" }
{ "_id" : 2, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 70, "items" : { "sku" : "chocolates", "qty" : 5, "price" : 10 }, "status" : "A" }
{ "_id" : 3, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 50, "items" : { "sku" : "oranges", "qty" : 10, "price" : 2.5 }, "status" : "A" }
{ "_id" : 3, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 50, "items" : { "sku" : "pears", "qty" : 10, "price" : 2.5 }, "status" : "A" }
{ "_id" : 4, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-18T00:00:00Z"), "price" : 25, "items" : { "sku" : "oranges", "qty" : 10, "price" : 2.5 }, "status" : "A" }
{ "_id" : 5, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-19T00:00:00Z"), "price" : 50, "items" : { "sku" : "chocolates", "qty" : 5, "price" : 10 }, "status" : "A" }
...

$group 阶段按 items.sku 分组，针对每个 sku 进行计算：

qty 字段。qty 字段包含
每个 items.sku 的订购 qty 总计（参见 $sum）。
orders_ids 数组。orders_ids 字段包含一个
items.sku 的不同顺序 _id 的数组（参见 $addToSet）。

{ "_id" : "chocolates", "qty" : 15, "orders_ids" : [ 2, 5, 8 ] }
{ "_id" : "oranges", "qty" : 63, "orders_ids" : [ 4, 7, 3, 2, 9, 1, 10 ] }
{ "_id" : "carrots", "qty" : 15, "orders_ids" : [ 6, 9 ] }
{ "_id" : "apples", "qty" : 35, "orders_ids" : [ 9, 8, 1, 6 ] }
{ "_id" : "pears", "qty" : 10, "orders_ids" : [ 3 ] }

$project 阶段会重塑输出文档以镜像 Map-Reduce 的输出，使其具有两个字段 _id 和 value。$project 会：

$unwind 阶段按 items 数组字段对文档进行分解，为每个数组元素输出一个文档。例如：

{ "_id" : 1, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-01T00:00:00Z"), "price" : 25, "items" : { "sku" : "oranges", "qty" : 5, "price" : 2.5 }, "status" : "A" }
{ "_id" : 1, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-01T00:00:00Z"), "price" : 25, "items" : { "sku" : "apples", "qty" : 5, "price" : 2.5 }, "status" : "A" }
{ "_id" : 2, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 70, "items" : { "sku" : "oranges", "qty" : 8, "price" : 2.5 }, "status" : "A" }
{ "_id" : 2, "cust_id" : "Ant O. Knee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 70, "items" : { "sku" : "chocolates", "qty" : 5, "price" : 10 }, "status" : "A" }
{ "_id" : 3, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 50, "items" : { "sku" : "oranges", "qty" : 10, "price" : 2.5 }, "status" : "A" }
{ "_id" : 3, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-08T00:00:00Z"), "price" : 50, "items" : { "sku" : "pears", "qty" : 10, "price" : 2.5 }, "status" : "A" }
{ "_id" : 4, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-18T00:00:00Z"), "price" : 25, "items" : { "sku" : "oranges", "qty" : 10, "price" : 2.5 }, "status" : "A" }
{ "_id" : 5, "cust_id" : "Busby Bee", "ord_date" : ISODate("2020-03-19T00:00:00Z"), "price" : 50, "items" : { "sku" : "chocolates", "qty" : 5, "price" : 10 }, "status" : "A" }
...

$group 阶段按 items.sku 分组，针对每个 sku 进行计算：

qty 字段。qty 字段包含使用 $sum 为每个 items.sku 订购的 qty 总计。
orders_ids 数组。orders_ids 字段包含使用 $addToSet 的 items.sku 的不同顺序 _id 的数组。

{ "_id" : "chocolates", "qty" : 15, "orders_ids" : [ 2, 5, 8 ] }
{ "_id" : "oranges", "qty" : 63, "orders_ids" : [ 4, 7, 3, 2, 9, 1, 10 ] }
{ "_id" : "carrots", "qty" : 15, "orders_ids" : [ 6, 9 ] }
{ "_id" : "apples", "qty" : 35, "orders_ids" : [ 9, 8, 1, 6 ] }
{ "_id" : "pears", "qty" : 10, "orders_ids" : [ 3 ] }

$project 阶段会重塑输出文档以镜像 Map-Reduce 的输出，使其具有两个字段 _id 和 value。$project 会：

使用 $size 将 value.count 设置为 orders_ids 数组的大小
将 value.qty 设置为输入文档中的 qty 字段。
使用 $divide 和 $size 将 value.avg 设置为每个订单的平均数量

{ "_id" : "apples", "value" : { "count" : 4, "qty" : 35, "avg" : 8.75 } }
{ "_id" : "pears", "value" : { "count" : 1, "qty" : 10, "avg" : 10 } }
{ "_id" : "chocolates", "value" : { "count" : 3, "qty" : 15, "avg" : 5 } }
{ "_id" : "oranges", "value" : { "count" : 7, "qty" : 63, "avg" : 9 } }
{ "_id" : "carrots", "value" : { "count" : 2, "qty" : 15, "avg" : 7.5 } }

最后，$merge 将输出写入集合 agg_alternative_3。如果现有文档的键 _id 与新结果相同，则操作会覆盖现有文档。如果没有具有相同键的现有文档，操作将插入该文档。

查询 agg_alternative_3 集合以验证结果：

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

该操作将返回以下文档：

{ "_id" : "apples", "value" : { "count" : 4, "qty" : 35, "avg" : 8.75 } }
{ "_id" : "carrots", "value" : { "count" : 2, "qty" : 15, "avg" : 7.5 } }
{ "_id" : "chocolates", "value" : { "count" : 3, "qty" : 15, "avg" : 5 } }
{ "_id" : "oranges", "value" : { "count" : 7, "qty" : 63, "avg" : 9 } }
{ "_id" : "pears", "value" : { "count" : 1, "qty" : 10, "avg" : 10 } }

提示

有关使用自定义聚合表达式的替代方案，请参阅 Map-Reduce 到聚合管道转换示例。

提示

后退

并发

来年

以增量执行