MongoDB学习笔记——聚合操作之MapReduce

时间:2023-03-08 17:10:16
MongoDB学习笔记——聚合操作之MapReduce

MapReduce

MongoDB中的MapReduce相当于关系数据库中的group by。使用MapReduce要实现两个函数Map和Reduce函数。Map函数调用emit(key,value),遍历 Collection中所有的记录,将key与value传递给Reduce函数进行处理。 Mapreduce使用惯用的javascript操作来做map和reduce操作,因此Mapreduce的灵活性和复杂性都会比aggregate 更高一些,并且相对aggregate 而言更消耗性能;

语法格式:

db.runCommand(

{

mapReduce: <collection>,

map: <function>,

reduce: <function>,

finalize: <function>,

out: <output>,

query: <document>,

sort: <document>,

limit: <number>,

scope: <document>,

jsMode: <boolean>,

verbose: <boolean>

}

)

等同于语法

db.collection.mapReduce(

<map>,

<reduce>,

{

out: <collection>,

query: <document>,

sort: <document>,

limit: <number>,

finalize: <function>,

scope: <document>,

jsMode: <boolean>,

verbose: <boolean>

})

参数说明:

  • mapReduce:要操作的目标集合
  • map:映射函数 (生成键值对序列,作为 reduce 函数参数), Map方法使用this来操作当前对象,至少调用一次 emit(key, value)方法向reduce提供参数。其中的key为最终结果集中的_id
  • reduce:统计函数,该函数接受map函数传来的key和value值。reduce函数中的key就是emit(key,value)中的key,而value是emit函数中同一个key返回的value数组。
  • query:一个筛选条件,只有满足条件的文档才会调用map函数。(query。limit,sort可以随意组合)
  • sort : 和limit结合的sort排序参数(也是在发往map函数前给文档排序),可以优化分组机制,提升mapreduce性能,
    处理未排序的集合意味着MapReduce引擎将得到随机顺序的值,在RAM中根本无法reduce。相反,它将不得不把所有文章写入一个临时收集的磁盘,然后按顺序读取并reduce。
  • limit: 发往map函数的文档数量的上限(要是没有limit,单独使用sort的用处不大)
  • finalize: 最终处理函数(对reduce返回结果进行最终整理后存入结果集合) finalize函数可能会在Reduce函数结束之后运行,这个函数是可选的,对于很多Map/Reduce任务来说不是必需的。finalize函数接收一个key和一个value,返回一个最终的value. 针对一个对象你的Reduce函数可能被调用了多次。当最后只需针对一个对象进行一次操作时可以使用finalize函数,比如计算平均值。
  • scope:向map、reduce、finalize导入外部变量
  • verbose : 指定是否在结果信息中包含的计时信息,默认true
  • jsMode: 布尔值,是否减少执行过程中BSON和JS的转换,默认false 对于MongoDB2.0及以上的版本,通常Map/Reduce的执行遵循下面两个步骤: a.从BSON转化为JSON,执行Map过程,将JSON转化为BOSN b.从BSON转化为JSON,执行Reduce过程,将JSON转化为BSON 因此,需要多次转化格式,但是可以利用临时集合在Map阶段处理很大的数据集。为了节省时间,可以利用{jsMode:ture}使Map/Reduce的执行保持在JSON状态。遵循如下两个步骤: a.从BSON转化为JSON,执行Map过程 b.执行Reduce过程,从 JSON转化为BSON 这样,执行时间可以大大减小,但需要注意,jsMode 受到JSON堆大小和独立主键最大500KB的限制。因此,对于较大的任务jsMode并不适用,在这种情况下会转变为通常的模式。
  • out:统计结果存放集合 (必填),
    在MongoDB1.8之前的版本,如果你没有指定out的值,那么结果将会被放到一个临时集合中,集合的名字在输出指令中指定,否则,你可以指定一个集合的名字作为out的选项,而结果将会被存储到你指定的集合中。
    对于MongoDB1.8以及以后的版本,输出选项改变了。Map/Reduce 不再产生临时集合,你必须为out指定一个值,设置out指令如下:

    out参数格式:

    out: { <action>: <collectionName>

    [, db: <dbName>]

    [, sharded: <boolean> ]

    [, nonAtomic: <boolean> ] }

    out参数说明

    • Action可以为 replace(默认)、merge、reduce {replace:"collectionName"}:输出结果将被插入到一个集合中,并且会自动替换掉现有的同名集合。该选项为默认的。 {merge:"collectionName"}:这个选项将会把新的数据连接到旧的输出结合中。换句话说,如果在结果集和旧集合中存在相同键值,那么新的键将会被覆盖掉。 {reduce:"collectionName"}:如果具有某个键值的文档同时存在于结果集和旧集合中,那么一个Reduce操作(利用特定的reduce函数)将作用于这个两个值,并且结果将会被写到输出集合中。如果指定了finalize函数,那么当Reduce结束后它将被执行。
    • db: 指明接收输出结果的数据库名称 out:{replace:"collectionName",db:"otherDB"}
    • shard: {shared:true}:适用于MongoDB1.9及以上的版本。如果设置为true,并且设置了数据库分片,那么输出的collection将被进行分片,并选择_id作为其片键。

MapReduce执行聚合的步骤

  • 1.执行query操作,针对想要聚合的集合进行数据筛选,只有满足条件的文档才会被继续执行
  • 2.执行sort操作,对满足条件的数据进行排序,可以优化分组的机制,通常与limit一起使用
  • 3.执行limit操作,对已经排序的数据进行过滤,筛选出能够执行map函数的文档上限,(要是没有limit,单独使用sort的用处不大)
  • 4.执行map操作,通过变量this来检验当前考察的对象,调用 emit(key, value)生成键值对序列,作为 reduce 函数参数
  • 5.执行reduce操作,处理需要统计的字段
  • 6.执行finalize操作,对reduce的结果执行finalize方法进行处理
  • 7.执行out操作,将结果集进行输出
  • 8.断开连接,临时Collection删除或保留。

编写MapReduce程序

所有的map-reduce函数都是用JavaScript书写,然后在mongod实例进程上运行。在进行map-reduce操作的时候,MongoDB会将满足查询条件的文档进行map所定义的操作,map函数会产生( emit)键值型的数据。
如果某个键所对应的值有多个的话,会进行reduce的操作,最后将结果保存到一个集合中。通过定义一个finalize函数可以对reduce的结果做进一步的处理,比如:进行投影或者规范化输出、进一步的计算等。
当我们的key-values中的values集合过大,会被再切分成很多个小的key-values块,然后分别执行Reduce函数,再将多个块的结果组合成一个新的集合,作为Reduce函数的第二个参数,继续Reducer操作。可以预见,如果我们初始的values非常大,可能还会对第一次分块计算后组成的集合再次Reduce。这就类似于多阶的归并排序了。具体会有多少重,就看数据量了。
上面这一内部机制,我们不必非常了解,但我们必须了解这一机制会要求我们遵守的原则,那就是当我们书写Map函数时,emit的第二个参数形式是我们的Reduce函数的第二个参数,而Reduce函数的返回值,可能会作为新的输入参数再次执行Reduce操作,所以Reduce函数的返回值也需要和Reduce函数的第二个参数结构一致

首先在order集合中插入测试数据

db.order.insert([{

"_id": ObjectId("528312e716b20807b2152db5"),

"cust_id": "1",

"ord_date": ISODate("2013-11-13T16:00:00Z"),

"status": "A",

"price": 25,

"items": [

{

"sku": "mmm",

"qty": 5,

"price": 2.5

},

{

"sku": "nnn",

"qty": 5,

"price": 2.5

}

]

},{

"_id": ObjectId("528312f716b20807b2152db6"),

"cust_id": "2",

"ord_date": ISODate("2013-11-13T16:00:00Z"),

"status": "A",

"price": 25,

"items": [

{

"sku": "mmm",

"qty": 5,

"price": 2.5

},

{

"sku": "nnn",

"qty": 5,

"price": 2.5

}

]

},{

"_id": ObjectId("5283130816b20807b2152db7"),

"cust_id": "3",

"ord_date": ISODate("2013-11-13T16:00:00Z"),

"status": "A",

"price": 25,

"items": [

{

"sku": "mmm",

"qty": 5,

"price": 2.5

},

{

"sku": "nnn",

"qty": 5,

"price": 2.5

}

]

},{

"_id": ObjectId("5283132c16b20807b2152db8"),

"cust_id": "3",

"ord_date": ISODate("2013-11-13T16:00:00Z"),

"status": "A",

"price": 30,

"items": [

{

"sku": "mmm",

"qty": 6,

"price": 2.5

},

{

"sku": "nnn",

"qty": 6,

"price": 2.5

}

]

},{

"_id": ObjectId("5283134d16b20807b2152db9"),

"cust_id": "2",

"ord_date": ISODate("2013-11-13T16:00:00Z"),

"status": "A",

"price": 20,

"items": [

{

"sku": "mmm",

"qty": 4,

"price": 2.5

},

{

"sku": "nnn",

"qty": 4,

"price": 2.5

}

]

}])

 

统计每个顾客的消费总金额

 

var mapFunc = function () {

emit(this.cust_id, this.price);

}

var reduceFunc = function (key, values) {

return Array.sum(values);

}

db.order.mapReduce(mapFunc, reduceFunc, { out: 'ordermapreduce' })

统计每种商品的购买次数和平均每次购买数量

var mapFunc = function () {

for (var i = 0; i < this.items.length; i++) {

var key = this.items[i].sku;

var value = { count: 1, qty: this.items[i].qty }

emit(key, value);

}

};

var reduceFunc = function (key, values) {

var result = { count: 0, qty: 0 };

for (var i = 0; i < values.length; i++) {

result.count += values[0].count;

result.qty += values[0].qty;

}

return result;

}

var finalizeFunc = function (key, reduceVal) {

reduceVal.avg = reduceVal.qty / reduceVal.count;

return reduceVal;

};

db.order.mapReduce(mapFunc, reduceFunc, { out: { merge: "ordermapreduce1" }, finalize: finalizeFunc });