详解Mongodb 多文档聚合操作处理方法(Map-reduce 函数)_Nosql

聚合

聚合操作处理多个文档并返回计算结果。您可以使用聚合操作来：

将多个文档中的值分组在一起。
对分组数据执行操作以返回单个结果。
分析数据随时间的变化。

要执行聚合操作，您可以使用：

聚合管道
单一目的聚合方法
map-reduce 函数

map-reduce 函数

在mongoshell 中，该db.collection.mapreduce() 方法是命令的包装器mapreduce。下面的例子使用该db.collection.mapreduce()方法。

定义： db.collection.mapreduce(map,reduce, { <options> })

该map功能有以下要求：

在map函数中，将当前文档引用为函数中的this。
该map函数不应出于任何原因访问数据库。
该map函数应该是纯粹的，或者对函数之外没有影响（即副作用）。
该map函数可以选择调用emit(key,value)任意次数来创建key与关联的输出文档value。

# 原型如下：
function() {
   ...
   emit(key, value);
}

该reduce函数表现出以下行为：

该reduce函数不应访问数据库，即使是执行读取操作。
该reduce功能不应影响外部系统。
reducemongodb 可以针对同一个键多次调用该函数。在这种情况下，该键的函数的先前输出将成为该键的reduce 下一个函数调用的输入值之一。
该reduce函数可以访问参数中定义的变量scope。

# 该reduce函数具有以下原型：
function(key, values) {
   ...
   return result;
}

插入测试数据。如下：

sit_rs1:primary> db.orders.insertmany([
...    { _id: 1, cust_id: "a", ord_date: new date("2023-06-01"), price: 15, items: [ { sku: "apple", qty: 5, price: 2.5 }, { sku: "apples", qty: 5, price: 2.5 } ], status: "1" },
...    { _id: 2, cust_id: "a", ord_date: new date("2023-06-08"), price: 60, items: [ { sku: "apple", qty: 8, price: 2.5 }, { sku: "banana", qty: 5, price: 10 } ], status: "1" },
...    { _id: 3, cust_id: "b", ord_date: new date("2023-06-08"), price: 55, items: [ { sku: "apple", qty: 10, price: 2.5 }, { sku: "pears", qty: 10, price: 2.5 } ], status: "1" },
...    { _id: 4, cust_id: "b", ord_date: new date("2023-06-18"), price: 26, items: [ { sku: "apple", qty: 10, price: 2.5 } ], status: "1" },
...    { _id: 5, cust_id: "b", ord_date: new date("2023-06-19"), price: 40, items: [ { sku: "banana", qty: 5, price: 10 } ], status: "1"},
...    { _id: 6, cust_id: "c", ord_date: new date("2023-06-19"), price: 38, items: [ { sku: "carrots", qty: 10, price: 1.0 }, { sku: "apples", qty: 10, price: 2.5 } ], status: "1" },
...    { _id: 7, cust_id: "c", ord_date: new date("2023-06-20"), price: 21, items: [ { sku: "apple", qty: 10, price: 2.5 } ], status: "1" },
...    { _id: 8, cust_id: "d", ord_date: new date("2023-06-20"), price: 76, items: [ { sku: "banana", qty: 5, price: 10 }, { sku: "apples", qty: 10, price: 2.5 } ], status: "1" },
...    { _id: 9, cust_id: "d", ord_date: new date("2023-06-20"), price: 51, items: [ { sku: "carrots", qty: 5, price: 1.0 }, { sku: "apples", qty: 10, price: 2.5 }, { sku: "apple", qty: 10, price: 2.5 } ], status: "1" },
...    { _id: 10, cust_id: "d", ord_date: new date("2023-06-23"), price: 23, items: [ { sku: "apple", qty: 10, price: 2.5 } ], status: "1" }
... ])
{
        "acknowledged" : true,
        "insertedids" : [
                1,
                2,
                3,
                4,
                5,
                6,
                7,
                8,
                9,
                10
        ]
}
sit_rs1:primary> db.orders.find()
{ "_id" : 4, "cust_id" : "b", "ord_date" : isodate("2023-06-18t00:00:00z"), "price" : 26, "items" : [ { "sku" : "apple", "qty" : 10, "price" : 2.5 } ], "status" : "1" }
{ "_id" : 6, "cust_id" : "c", "ord_date" : isodate("2023-06-19t00:00:00z"), "price" : 38, "items" : [ { "sku" : "carrots", "qty" : 10, "price" : 1 }, { "sku" : "apples", "qty" : 10, "price" : 2.5 } ], "status" : "1" }
{ "_id" : 1, "cust_id" : "a", "ord_date" : isodate("2023-06-01t00:00:00z"), "price" : 15, "items" : [ { "sku" : "apple", "qty" : 5, "price" : 2.5 }, { "sku" : "apples", "qty" : 5, "price" : 2.5 } ], "status" : "1" }
{ "_id" : 2, "cust_id" : "a", "ord_date" : isodate("2023-06-08t00:00:00z"), "price" : 60, "items" : [ { "sku" : "apple", "qty" : 8, "price" : 2.5 }, { "sku" : "banana", "qty" : 5, "price" : 10 } ], "status" : "1" }
{ "_id" : 9, "cust_id" : "d", "ord_date" : isodate("2023-06-20t00:00:00z"), "price" : 51, "items" : [ { "sku" : "carrots", "qty" : 5, "price" : 1 }, { "sku" : "apples", "qty" : 10, "price" : 2.5 }, { "sku" : "apple", "qty" : 10, "price" : 2.5 } ], "status" : "1" }
{ "_id" : 3, "cust_id" : "b", "ord_date" : isodate("2023-06-08t00:00:00z"), "price" : 55, "items" : [ { "sku" : "apple", "qty" : 10, "price" : 2.5 }, { "sku" : "pears", "qty" : 10, "price" : 2.5 } ], "status" : "1" }
{ "_id" : 5, "cust_id" : "b", "ord_date" : isodate("2023-06-19t00:00:00z"), "price" : 40, "items" : [ { "sku" : "banana", "qty" : 5, "price" : 10 } ], "status" : "1" }
{ "_id" : 7, "cust_id" : "c", "ord_date" : isodate("2023-06-20t00:00:00z"), "price" : 21, "items" : [ { "sku" : "apple", "qty" : 10, "price" : 2.5 } ], "status" : "1" }
{ "_id" : 8, "cust_id" : "d", "ord_date" : isodate("2023-06-20t00:00:00z"), "price" : 76, "items" : [ { "sku" : "banana", "qty" : 5, "price" : 10 }, { "sku" : "apples", "qty" : 10, "price" : 2.5 } ], "status" : "1" }
{ "_id" : 10, "cust_id" : "d", "ord_date" : isodate("2023-06-23t00:00:00z"), "price" : 23, "items" : [ { "sku" : "apple", "qty" : 10, "price" : 2.5 } ], "status" : "1" }

示例：按客户统计

对集合 orders 执行map-reduce操作，按 cust_id 进行分组，然后统计每个客户的 price 计算总和，如下：

首先，我们需要定义map函数来处理每个输入文档:

在函数中，this指的是map-reduce操作正在处理的文档。
该函数将每个文档的 price 映射为 cust_id，并发出 cust_id 和 price 。

sit_rs1:primary> var mymapfun = function() {
...    emit(this.cust_id, this.price);
... };
sit_rs1:primary> print(mymapfun)
function() {
   emit(this.cust_id, this.price);
}

然后，用两个参数 keycustid 和 valuesprices 定义相应的reduce函数。这里需要调用数组的 sum 方法计算客户订单总价。

valuesprices 是一个数组，其元素是map函数发出的price 字段的值，并按 keycustid 分组。
该函数将 valuesprice 数组缩减为其元素的总和

# 计算数组元素总和
sit_rs1:primary> array.sum([2,2,6,8])
18
# 计算数组平均值
sit_rs1:primary> array.avg([1,2,3])
2
sit_rs1:primary> var myreducefun = function(keycustid, valuesprices) {
...    return array.sum(valuesprices);
... };
sit_rs1:primary> print(myreducefun)
function(keycustid, valuesprices) {
   return array.sum(valuesprices);
}

最后，使用 mymapfun 函数和 myreducefun 函数对集合 orders 中的所有文档执行map-reduce统计：

out: 指定map-reduce操作结果的位置。您可以输出到集合、通过操作输出到集合或内联输出。
此操作将结果输出到名为的集合 map_reduce_out。如果该 map_reduce_out 集合已存在，则该操作将使用此 map-reduce 操作的结果替换内容。

sit_rs1:primary> db.orders.mapreduce(
...    mymapfun,
...    myreducefun,
...    { out: "map_reduce_out" }
... )
{
        "result" : "map_reduce_out",
        "ok" : 1,
        "$clustertime" : {
                "clustertime" : timestamp(1690259241, 6),
                "signature" : {
                        "hash" : bindata(0,"kur+uesljyct5oexd8ujpic/j3q="),
                        "keyid" : numberlong("7205479298910650370")
                }
        },
        "operationtime" : timestamp(1690259241, 6)
}

查询 map_reduce_out 集合以验证结果是否正确：

sit_rs1:primary> db.map_reduce_out.find().sort( { _id: 1 } )
{ "_id" : "a", "value" : 75 }
{ "_id" : "b", "value" : 121 }
{ "_id" : "c", "value" : 59 }
{ "_id" : "d", "value" : 150 }
# 检查 cust_id 为 a 的客户， 总和是 75 正确
sit_rs1:primary> db.orders.find({ "cust_id" : "a"}, {"price": 1})
{ "_id" : 1, "price" : 15 }
{ "_id" : 2, "price" : 60 }
# 检查 cust_id 为 b 的客户，总和是 121 正确
sit_rs1:primary> db.orders.find({ "cust_id" : "b"}, {"price": 1})
{ "_id" : 4, "price" : 26 }
{ "_id" : 3, "price" : 55 }
{ "_id" : 5, "price" : 40 }

示例：按日期统计

按日期统计，和上面示例一样，只需要把 map 函数重新定义如下，将每个文档的 price 映射为 ord_date，并发出 ord_date 和 price 。

sit_rs1:primary> var mymapfun2 = function() {
...     emit(this.ord_date, this.price);
... };
sit_rs1:primary> print(mymapfun2)
function() {
    emit(this.ord_date, this.price);
}

然后，用两个参数 keyorddate 和 valuesprices 定义相应的reduce函数。这里需要调用数组的 avg 方法计算平均客单价。

valuesprices 是一个数组，其元素是map函数发出的 price 字段的值，并按 keyorddate 分组。
该函数将 valuesprice 数组缩减为其元素的总和的平均值

sit_rs1:primary> var myreducefun2 = function(keyorddate, valuesprices) {
...    return array.avg(valuesprices);
... };
sit_rs1:primary> print(myreducefun2)
function(keyorddate, valuesprices) {
   return array.avg(valuesprices);
}

最后，使用 mymapfun2 函数和 myreducefun2 函数对集合 orders 中的所有文档执行map-reduce统计：

sit_rs1:primary> db.orders.mapreduce(
...    mymapfun2,
...    myreducefun2,
...    { out: "map_reduce_out2" }
... )
{
        "result" : "map_reduce_out2",
        "ok" : 1,
        "$clustertime" : {
                "clustertime" : timestamp(1690265083, 8),
                "signature" : {
                        "hash" : bindata(0,"pcwsky3hjlgejsk00arydzkecde="),
                        "keyid" : numberlong("7205479298910650370")
                }
        },
        "operationtime" : timestamp(1690265083, 8)
}

查询 map_reduce_out2 集合以验证结果是否正确：

sit_rs1:primary> db.map_reduce_out2.find()
{ "_id" : isodate("2023-06-08t00:00:00z"), "value" : 57.5 }
{ "_id" : isodate("2023-06-01t00:00:00z"), "value" : 15 }
{ "_id" : isodate("2023-06-18t00:00:00z"), "value" : 26 }
{ "_id" : isodate("2023-06-20t00:00:00z"), "value" : 49.333333333333336 }
{ "_id" : isodate("2023-06-23t00:00:00z"), "value" : 23 }
{ "_id" : isodate("2023-06-19t00:00:00z"), "value" : 39 }
# 检查日期2023-06-08的订单平均值
sit_rs1:primary> db.orders.find({ "ord_date" : isodate("2023-06-08t00:00:00z")}, {"price": 1})
{ "_id" : 2, "price" : 60 }
{ "_id" : 3, "price" : 55 }
sit_rs1:primary> print((60+55)/2)
57.5
# 检查日期2023-06-20的订单平均值
sit_rs1:primary> db.orders.find({ "ord_date" : isodate("2023-06-20t00:00:00z")}, {"price": 1})
{ "_id" : 9, "price" : 51 }
{ "_id" : 7, "price" : 21 }
{ "_id" : 8, "price" : 76 }
sit_rs1:primary> print((51+21+76)/3)
49.333333333333336

对于需要自定义功能的 map-reduce 操作，mongodb 从 4.4 版本开始提供 $accumulator 和 $function 聚合运算符。使用这些运算符在 javascript 中自定义聚合表达式。