Map-Reduce 示例
注意
作为 Map-Reduce 替代方案的聚合管道
在 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 } 查询
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 }
通过每款商品的平均数量计算订单和总数量
在以下示例中,您将看到针对 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
字段。
{ "_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 } }