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 });