MongoDB：19-MongoDB-Map Reduce

1. Map-Reduce是一种计算模型，简单的说就是将大批量的工作（数据）分解（MAP）执行，然后再将结果合并成最终结果（REDUCE）。
3. MongoDB提供的Map-Reduce非常灵活，对于大规模数据分析也相当实用。

MapReduce 命令

• 以下是MapReduce的基本语法：

1. >db.collection.mapReduce(
2.   function() {emit(key,value);},  //map 函数
3.   function(key,values) {return reduceFunction},   //reduce 函数
4.   {
5.      out: collection,
6.      query: document,
7.      sort: document,
8.      limit: number
9.   }
10. )

11. db.collection.mapReduce(
12.                         <map>,
13.                         <reduce>,
14.                         {
15.                           out: <collection>,
16.                           query: <document>,
17.                           sort: <document>,
18.                           limit: <number>,
19.                           finalize: <function>,
20.                           scope: <document>,
21.                           jsMode: <boolean>,
22.                           verbose: <boolean>,
23.                           bypassDocumentValidation: <boolean>
24.                         }
25.                       )

26. MapReduce要实现两个函数：Map和Reduce。
28. Map函数调用emit(key,value)遍历一个或多个集合中所有的记录,进行分组（group by），
29. 然后将key与value传给Reduce函数进行处理，输出结果。
30.  
31. （1）MapReduce使用自定义JavaScript函数执行map和reduce操作，所以是基于js引擎，单线程执行，效率不高，
32.     比Aggregation复杂，适合用做后台统计等。
33. （2）MapReduce支持分片操作，可以进行拆分，分发到不同的机器上执行（多服务器并行做数据集合处理），
34.     然后再将不同的机器处理的结果汇集起 来，输出结果,。
35. （3）MapReduce能执行单一聚合的所有操作count、distinct、group，
36.     但group 在当数据量非常大的时候，处理能力就不太好，先筛选再分组，不支持 分片，对数据量有所限制，效率不高。

参数说明:

1. 参数说明：
2.    map:是JavaScript 函数，负责将每一个输入文档转换为零或多个文档，通过key进行分组，生成键值对序列,作为 reduce 函数参数。
3.    reduce：是JavaScript 函数，对map操作的输出做合并的化简的操作（将key-values变成key-value，也就是把values数组变成一个单一的值value）。
4.    out：reduce执行完，存放的集合，如果不指定集合，则使用默认的临时集合，在MapReduce的连接关闭后自动就被删除了。

• out:{<action>:<collectionName>
• [, db:<dbName>]
• [, sharded:<boolean>]
• [, nonAtomic:<boolean>]}

5.    query：过滤的条件，对符合条件的文档执行map函数。（query。limit，sort可以随意组合）。
6.    sort ：对文档进行排序，sort和limit结合的sort排序参数（也是在发往map函数前给文档排序），可以优化分组机制。
7.    limit ：发往map函数的文档数量的上限（要是没有limit，单独使用sort的用处不大）。
8.  finalize：可以对reduce输出结果再一次修改，跟group的finalize一样，不过MapReduce没有group的4MB文档的输出限制。
9.  scope：向map、reduce、finalize导入外部变量。
10.  verbose:是否包括结果信息中的时间信息，默认为fasle。
11.     "timing" : {
12.                "mapTime" : 0,
13.                "emitLoop" : 2,
14.                "reduceTime" : 0,
15.                "mode" : "mixed",
16.                "total" : 0
17.        }

以下实例在集合 orders 中查找 status:"A" 的数据，并根据 cust_id 来分组，并计算 amount 的总和。

使用 MapReduce

• 考虑以下文档结构存储用户的文章，文档存储了用户的 user_name 和文章的 status 字段：
  

1. db.posts.insert([
2. {
3.   "post_text": "Spring",
4.   "user_name": "fly",
5.   "status":"active"
6. },
7. {
8.   "post_text": "Struts2",
9.   "user_name": "fly",
10.   "status":"active"
11. },
12. {
13.   "post_text": "Hibernate",
14.   "user_name": "fly",
15.   "status":"disabled"
16. },
17. {
18.   "post_text": "Mybatis",
19.   "user_name": "fly",
20.   "status":"active"
21. },
22. {
23.   "post_text": "Redis",
24.   "user_name": "fly",
25.   "status":"disabled"
26. },
27. {
28.   "post_text": "Restful",
29.   "user_name": "cc",
30.   "status":"active"
31. },
32. {
33.   "post_text": "Dubbo",
34.   "user_name": "cc",
35.   "status":"disabled"
36. },
37. {
38.   "post_text": "SpringCloud",
39.   "user_name": "cc",
40.   "status":"active"
41. },
42. {
43.   "post_text": "SpringBoot",
44.   "user_name": "cc",
45.   "status":"disabled"
46. },
47. {
48.   "post_text": "MongoDB",
49.   "user_name": "cc",
50.   "status":"active"
51. },
52. {
53.   "post_text": "MySQL",
54.   "user_name": "cc",
55.   "status":"active"
56. }
57. ])

• 现在，我们将在 posts 集合中使用 mapReduce 函数来选取已发布的文章(status:"active")，并通过user_name分组，计算每个用户的文章数：

1. >db.posts.mapReduce(
2.   function() { emit(this.user_name,1); },
3.   function(key, values) {return Array.sum(values)},
4.      {  
5.         query:{status:"active"},  
6.         out:"post_total"
7.      }
8. )

• 以上 mapReduce 输出结果为：

1. /* 1 */
2. {
3.    "result" : "post_total",
4. "timeMillis" : 409.0,
5. "counts" : {
6. "input" : 7,
7. "emit" : 7,
8. "reduce" : 2,
9. "output" : 2
10. },
11.    "ok" : 1.0,
12.    "_o" : {
13.        "result" : "post_total",
14.        "timeMillis" : 409,
15.        "counts" : {
16.            "input" : 7,
17.            "emit" : 7,
18.            "reduce" : 2,
19.            "output" : 2
20.        },
21.        "ok" : 1.0
22.    },
23.    "_keys" : [
24.        "result",
25.        "timeMillis",
26.        "counts",
27.        "ok"
28.    ],
29.    "_db" : {
30.        "_mongo" : {
31.            "slaveOk" : true,
32.            "host" : "localhost:27017",
33.            "defaultDB" : "mongotest",
34.            "authStatus" : {
35.                "authRequired" : true,
36.                "isMaster" : true,
37.                "replSetGetStatus" : true
38.            },
39.            "_readMode" : "commands",
40.            "_writeMode" : "commands"
41.        },
42.        "_name" : "mongotest"
43.    },
44.    "_coll" : {
45.        "_mongo" : {
46.            "slaveOk" : true,
47.            "host" : "localhost:27017",
48.            "defaultDB" : "mongotest",
49.            "authStatus" : {
50.                "authRequired" : true,
51.                "isMaster" : true,
52.                "replSetGetStatus" : true
53.            },
54.            "_readMode" : "commands",
55.            "_writeMode" : "commands"
56.        },
57.        "_db" : {
58.            "_mongo" : {
59.                "slaveOk" : true,
60.                "host" : "localhost:27017",
61.                "defaultDB" : "mongotest",
62.                "authStatus" : {
63.                    "authRequired" : true,
64.                    "isMaster" : true,
65.                    "replSetGetStatus" : true
66.                },
67.                "_readMode" : "commands",
68.                "_writeMode" : "commands"
69.            },
70.            "_name" : "mongotest"
71.        },
72.        "_shortName" : "post_total",
73.        "_fullName" : "mongotest.post_total"
74.    }
75. }

• 结果表明，共有 7个符合查询条件（status:"active"）的文档， 在map函数中生成了 7个键值对文档，最后使用reduce函数将相同的键值分为 2 组。
  
• 具体参数说明：

1. result：reduce执行完，存放的集合，如果不指定集合，则使用默认的临时集合，在MapReduce的连接关闭后自动就被删除了。
2. timeMillis：执行花费的时间，毫秒为单位
3. input：满足条件被发送到map函数的文档个数
4. emit：在map函数中emit被调用的次数，也就是所有集合中的数据总量
5. ouput：结果集合中的文档个数（count对调试非常有帮助）
6. ok：是否成功，成功为1
7. err：如果失败，这里可以有失败原因，不过从经验上来看，原因比较模糊，作用不大

使用 find 操作符来查看 mapReduce 的查询结果：

8. >db.posts.mapReduce(
9.   function() { emit(this.user_name,1); },
10.   function(key, values) {return Array.sum(values)},
11.      {  
12.         query:{status:"active"},  
13.         out:"post_total"
14.      }
15. ).find()

• 以上查询显示如下结果，两个用户 tom 和 mark 有两个发布的文章:
  

1. /* 1 */
2. {
3.    "_id" : "cc",
4.    "value" : 4.0
5. }
7. /* 2 */
8. {
9.    "_id" : "fly",
10.    "value" : 3.0
11. }

• 用类似的方式，MapReduce可以被用来构建大型复杂的聚合查询。
  
• Map函数和Reduce函数可以使用 JavaScript 来实现，使得MapReduce的使用非常灵活和强大。
  

附录：参数详解

• map函数

1. 【map函数】
2. map是JavaScript 函数,负责将每一个输入文档转换为零或多个文档，通过key进行分组，生成键值对序列,作为 reduce 函数参数。

1. function() {
2.       emit(key, value);
3.  }

3. 
   
4. key对文档进行分组，value是要统计的数据，value可以是JSON对象（emit只能容纳MongoDB的最大BSON文件大小的一半）。
6. 我们对订单的详细统计每个产品类型卖出了多少个。
8. 
   

• 订单数据    

9. db.item.insert( [  
10.  {  
11.   "quantity" : 2,  
12.   "price" : 5.0,  
13.   "pnumber" : "p003"  
14.  },{  
15.   "quantity" : 2,  
16.   "price" : 8.0,  
17.   "pnumber" : "p002"  
18.  },{  
19.   "quantity" : 1,  
20.   "price" : 4.0,  
21.   "pnumber" : "p002"  
22.  },{  
23.   "quantity" : 2,  
24.   "price" : 4.0,  
25.   "pnumber" : "p001"  
26.  },{  
27.   "quantity" : 4,  
28.   "price" : 10.0,  
29.   "pnumber" : "p003"  
30.  },{  
31.   "quantity" : 10,  
32.   "price" : 20.0,  
33.   "pnumber" : "p001"  
34.  },{  
35.   "quantity" : 10,  
36.   "price" : 20.0,  
37.   "pnumber" : "p003"  
38.  },{  
39.   "quantity" : 5,  
40.   "price" : 10.0,  
41.   "pnumber" : "p002"  
42.  }  
43. ])  
44. 我们先通过 pnumber进行分组，然后在对 quantity相加
45. 相当于select pnumber,sum(quantity) from item   group by pnumber
46. varmap= function() { emit(this.pnumber,this.quantity)}  
47. var reduce=function(key,values){return {'pumber':key,'quantity':Array.sum(values)}}  
48. db.item.mapReduce( map,  
49.                   reduce,  
50.                    { out: "map_reduce_data" }    
51.                   )  
52. db.map_reduce_data.find()  

• 输出结果

53. /* 1 */
54. {
55.    "_id" : "p001",
56.    "value" : {
57.        "pumber" : "p001",
58.        "quantity" : 12.0
59.    }
60. }
62. /* 2 */
63. {
64.    "_id" : "p002",
65.    "value" : {
66.        "pumber" : "p002",
67.        "quantity" : 8.0
68.    }
69. }
71. /* 3 */
72. {
73.    "_id" : "p003",
74.    "value" : {
75.        "pumber" : "p003",
76.        "quantity" : 16.0
77.    }
78. }

• query过滤的条件

1. 对符合条件的文档将会执行map函数。（query。limit，sort可以随意组合），
3. 我们对订单的详细的每次每种产品卖出的数量要大于5的并统计每个产品类型卖出了多少个。
5. 我们先通过 pnumber进行分组，然后在对 quantity相加
6. 相当于select pnumber,sum(quantity) from item where quantity>5   group by pnumber
7. varmap= function() { emit(this.pnumber,this.quantity)}  
8. var reduce=function(key,values){return {'pumber':key,'quantity':Array.sum(values)}}  
9. db.item.mapReduce( map,                        
10.                   reduce,                    
11.                    { query:{'quantity':{$gt:5}},  
12.                    out: "map_reduce_data" }      
13. )  
14. db.map_reduce_data.find()  
15. /* 1 */
16. {
17.    "_id" : "p001",
18.    "value" : 10.0
19. }
21. /* 2 */
22. {
23.    "_id" : "p003",
24.    "value" : 10.0
25. }

• value是JSON对象

26. value可以是JSON格式，我们对订单的详细统计每个产品类型出现次数。
28. 我们先通过 pnumber进行分组，然后在对 quantity相加 相当于select pnumber,count(*) from item   group by pnumber。
29. varmap= function() {emit(this.pnumber,{count:1});}  
30. var reduce=function(key,values){  
31.    var count=0;    
32.    values.forEach(function(val){ count+=val.count;});    
33.    return {'pumber':key,"count":count};  
34.  }  
35. db.item.mapReduce( map,  
36.                    reduce,  
37.                    { out: "map_reduce_data" }    
38. )  
39. db.map_reduce_data.find()  
40. /* 1 */
41. {
42.    "_id" : "p001",
43.    "value" : {
44.        "pumber" : "p001",
45.        "count" : 2.0
46.    }
47. }
49. /* 2 */
50. {
51.    "_id" : "p002",
52.    "value" : {
53.        "pumber" : "p002",
54.        "count" : 3.0
55.    }
56. }
58. /* 3 */
59. {
60.    "_id" : "p003",
61.    "value" : {
62.        "pumber" : "p003",
63.        "count" : 3.0
64.    }
65. }

• emit多次的循环

66. 可以对emit多次的循环，可以根据输入文档的项目字段中的元素的数量（键，值）多次调用：

1. function() {
2.        this.items.forEach(function(item){ emit(key, value); });
3.     }

• 数据：    

68. db.orders.insert( [  
69. {  
70. "onumber" : "001",  
71. "item" : [{  
72.   "quantity" : 2,  
73.   "price" : 5.0,  
74.   "pnumber" : "p003"  
75.  },{  
76.   "quantity" : 2,  
77.   "price" : 8.0,  
78.   "pnumber" : "p002"  
79.  }]  
80. },{  
81. "onumber" : "002",  
82. "item" : [{  
83.   "quantity" : 1,  
84.   "price" : 4.0,  
85.   "pnumber" : "p002"  
86.  },{  
87.   "quantity" : 2,  
88.   "price" : 4.0,  
89.   "pnumber" : "p001"  
90.  },{  
91.   "quantity" : 4,  
92.   "price" : 10.0,  
93.   "pnumber" : "p003"  
94.  }]  
95. },{  
96. "onumber" : "003",  
97. "item" : [{  
98.   "quantity" : 10,  
99.   "price" : 20.0,  
100.   "pnumber" : "p001"  
101.  },{  
102.   "quantity" : 10,  
103.   "price" : 20.0,  
104.   "pnumber" : "p003"  
105.  }]  
106. },{  
107. "onumber" : "004",  
108. "item" : [{  
109.   "quantity" : 5,  
110.   "price" : 10.0,  
111.   "pnumber" : "p002"  
112.  }]  
113. }  
114. ])    
115. 我们对统计订单中对应的产品销售了多少个，我们先通过 pnumber进行分组，然后在对 quantity相加。
116. varmap= function() { this.item.forEach(function(it){ emit(it.pnumber,it.quantity); })}  
117. var reduce=function(key,values){return {'pumber':key,'quantity':Array.sum(values)}}  
118. db.orders.mapReduce( map,                        
119.                     reduce,                    
120.                     { out: "map_reduce_data" }    
121. )

• 输出结果：

122. db.map_reduce_data.find()  
123. /* 1 */
124. {
125.    "_id" : "p001",
126.    "value" : {
127.        "pumber" : "p001",
128.        "quantity" : 12.0
129.    }
130. }
132. /* 2 */
133. {
134.    "_id" : "p002",
135.    "value" : {
136.        "pumber" : "p002",
137.        "quantity" : 8.0
138.    }
139. }
141. /* 3 */
142. {
143.    "_id" : "p003",
144.    "value" : {
145.        "pumber" : "p003",
146.        "quantity" : 16.0
147.    }
148. }
149. > varmap= function() { this.item.forEach(function(it){ emit(it.pnumber,it.quantity); })}
150. 也可以这样写
151. > varmap= function() { for(var i=0;i<this.item.length;i++){ var it=item[i]; emit(it.pnumber,it.quantity); }}

• reduce函数

1.  reduce是JavaScript 函数，对map操作的输出做合并的化简的操作
2. （将key-values变成key-value，也就是把values数组变成一个单一的值value）。  

1.    function(key, values) {
2.       ...
3.       return result;
4.    }

6.  values：值参数是一个数组,返回对象的类型必须与由map函数发出的值的类型相同。
7.  reduce：函数应该交换：即结果中元素的顺序不影响reduce函数的输出。

1. reduce( key, [ A, B ] ) == reduce( key, [ B, A ] )    

9. 对map操作的输出做合并的化简的操作（将key-values变成key-value，也就是把values数组变成一个单一的值value）,
10. 我们对订单的详细统计每个产品类型卖出了多少个和每种产品出现次数。
11. 我们先通过 pnumber进行分组，然后在对 quantity相加
12. 相当于select pnumber,count(*),sum(quantity) from item   group by pnumber
13. varmap= function() {  
14.     var value={count:1, quantity:this.quantity};  
15.    emit(this.pnumber,value);  
16.  }  
17. var reduce=function(key,values){  
18.                     varreducedVal= { count: 0, quantity: 0 };  
19.                     for (vari=0; i < values.length; i++) {  
20.                         reducedVal.count += values[i].count;  
21.                         reducedVal.quantity += values[i].quantity;  
22.                     }  
23.                     return reducedVal;  
24. }  
26. db.item.mapReduce( map,                        
27.                      reduce,                    
28.                      { out: "map_reduce_data" }      
29. ).find()  

• 输出结果：

30. /* 1 */
31. {
32.    "_id" : "p001",
33.    "value" : {
34.        "count" : 2.0,
35.        "quantity" : 12.0
36.    }
37. }
39. /* 2 */
40. {
41.    "_id" : "p002",
42.    "value" : {
43.        "count" : 3.0,
44.        "quantity" : 8.0
45.    }
46. }
48. /* 3 */
49. {
50.    "_id" : "p003",
51.    "value" : {
52.        "count" : 3.0,
53.        "quantity" : 16.0
54.    }
55. }

参考来源： http://www.runoob.com/mongodb/mongodb-map-reduce.html

参考来源：http://blog.csdn.net/congcong68/article/details/51471460