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