MongoDB Map-Reduce实例

在 mongo shell 中，db.collection.mapReduce() 方法是 mapReduce 命令的包装。下面的示例使用 db.collection.mapReduce() 方法：

准备数据

使用以下文档创建一个简单的 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 的价格之和：

（1）定义 map 函数以处理每个输入文档：

• 在函数中，this 表示 map-reduce 操作正在处理的文档。

• 该函数将每个文档的 price 映射到 cust_id，并发出 cust_id 和 price 对。

var mapFunction1 = function() {
   emit(this.cust_id, this.price);
};

（2）使用 keyCustId  和 valuesPrices 两个参数定义相应的 reduce 函数：

• valuesPrices 是一个数组，其元素是由 map 函数发出并按 keyCustId 分组的价格值。

• 该函数将 valuePrices 数组简化为其元素之和。

var reduceFunction1 = function(keyCustId, valuesPrices) {
   return Array.sum(valuesPrices);
};

（3）使用 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  集合的内容。

mongodb广告位

（4）查询 the 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

创建 demo1.js 文件，该文件内容如下：

// 定义 Map 函数
var mapFunction1 = function() {
    emit(this.cust_id, this.price);
};

// 定义 Reduce 函数
var reduceFunction1 = function(keyCustId, valuesPrices) {
    return Array.sum(valuesPrices);
};

// 执行 map-reduce 操作
db.orders.mapReduce(
    mapFunction1,
    reduceFunction1,
    { out: "map_reduce_example" }
);

// 查询结果集，验证结果
var examples = db.map_reduce_example.find();
while(examples.hasNext()) {
    printjson( examples.next() );
}

通过如下命令“.mongo.exe .demo1.js”去运行脚本，如下：

D:mongodb-v4.0.2-x86in> .mongo.exe .demo1.js                                                             
MongoDB shell version v4.0.2-143-g7ea530946f
connecting to: mongodb://127.0.0.1:27017
Implicit session: session { "id" : UUID("67347b05-d0d2-4b78-8b90-33d5a92a5a3d") }
MongoDB server version: 4.0.2-143-g7ea530946f
{ "_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" }
])

（1）$group 阶段按 cust_id 分组并计算字段值。value 字段包含每个 cust_id 的总价。该阶段将以下文档输出到下一阶段：

{ "_id" : "Don Quis", "value" : 155 }
{ "_id" : "Ant O. Knee", "value" : 95 }
{ "_id" : "Cam Elot", "value" : 60 }
{ "_id" : "Busby Bee", "value" : 125 }

（2）然后, $out 将输出结果输入到 collection agg_alternative_1 集合。或者，您可以使用 $merge 而不是 $out。

mongodb广告位

（3）查询 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 操作，如下：

C:UsersAdministrator> mongo
MongoDB shell version v4.0.2-143-g7ea530946f
connecting to: mongodb://127.0.0.1:27017
Implicit session: session { "id" : UUID("3a77e665-bfde-489b-bd7f-7d666f38cd43") }
MongoDB server version: 4.0.2-143-g7ea530946f
> use test
switched to db test
> db.auth("test","123456");
1
> db.orders.aggregate([
...    { $group: { _id: "$cust_id", value: { $sum: "$price" } } },
...    { $out: "agg_alternative_1" }
... ]);
> db.agg_alternative_1.find();
{ "_id" : "Busby Bee", "value" : 125 }
{ "_id" : "Ant O. Knee", "value" : 95 }
{ "_id" : "Don Quis", "value" : 155 }
{ "_id" : "Cam Elot", "value" : 60 }
{ "_id" : "B212", "value" : 0 }
{ "_id" : "A123", "value" : 0 }
>