MongoDB 关于Map及Reduce

     Hadoop中的map/reduce(分布式计算模型)

     Mapper:对输入的列表中的每一个元素执行一个函数,生成一个列表结果。(运算前后列表的元素数量不变)

     Reduce:对输入的列表的所有元素执行一个函数操作,并将中间结果作为参数继续执行函数。

    

     map把(k1,v1)变成(k2,v2),shuffle把(k2,v2)变成(k2,list(v2)),reduce把(k2,list(v2))变成(k3,v3)。

     (k1,v1)来自于输入，是已知的,shuffle过程由程序自动完成，(k3,v3)也是我们知道的,所以程序员要实现

     的是(k2,v2)。就是说我们知道输入的数据的格式及内容，也知道输出的格式，但是输出内容是由程序计算

     出来的，可以当做是已知的。

     总的来说就是处理输入，产生输出。先把大数据map(分解)，再逐一reduce(化简)，形成最终结果。

 

     MapReduce在MongoDB中主要用于批量处理数据和集合操作，类似于RDBMS的group操作(分组统计)。

     用this访问当前文档，map中必须有emit函数，才能把map结果传给reduce。

     emit(key,value)，key是分组的依据,value是你想要的东西。(也就是要统计的数据)

     如果map结果数据还是很大，可以再切分，然后分别执行reduce函数，所以reduce一定是可以反复利用。

     Map和Reduce函数用Js编写，语法遵循js标准。 

  

     以下是MongoDB官方的一个例子，就是统计一个标签系统中每个标签出现的次数。

    

> db.things.insert( { _id : 1, tags : ['dog', 'cat'] } );
> db.things.insert( { _id : 2, tags : ['cat'] } );
> db.things.insert( { _id : 3, tags : ['mouse', 'cat', 'dog'] } );
> db.things.insert( { _id : 4, tags : []  } );

> // map function
> map = function(){
...    this.tags.forEach(
...     function(z){//emit会遍历集合中的记录
...      emit( z , { count : 1 } ); //emit可以理解为调用reduce方法，这里参数为1[即累加1操作](统计)
...     }
...    );
...};

> // reduce function
> reduce = function( key , values ){
...    var total = 0;
...    for ( var i=0; i<values.length; i++ )
...        total += values[i].count;
...    return { count : total };
...};

//这个不属于官方例子，忽略掉
> f= function( key , rvalue ){
   //key 就是分组的依据
   //rvalue 就是reduce的结果 如value的一个实例 {count:6}
   if("cat"==key){
     rvalue.msg="this is cat tag ."
   }
   return rvalue;
...};

db.things.mapReduce(map,reduce,{out:'tmp'})
{
    "result" : "tmp",
    "timeMillis" : 316,
    "counts" : {
        "input" : 4,
        "emit" : 6,
        "output" : 3
    },
    "ok" : 1,
}
> db.tmp.find()
{ "_id" : "cat", "value" : { "count" : 3 } }
{ "_id" : "dog", "value" : { "count" : 2 } } // _id就是key , value就是处理后的返回值
{ "_id" : "mouse", "value" : { "count" : 1 } }

   

    MapReduce函数的参数列表如下:

   

db.runCommand(
 { mapreduce : <collection>,
   map : <mapfunction>,
   reduce : <reducefunction>
   [, query : <query filter object>]
   [, sort : <sort the query.  useful for optimization>]
   [, limit : <number of objects to return from collection>]
   [, out : <output-collection name>]
   [, keeptemp: <true|false>]
   [, finalize : <finalizefunction>]
   [, scope : <object where fields go into javascript global scope >]
   [, verbose : true]
 }
);
或者这么写：
db.collection.mapReduce(
                         <map>,
                         <reduce>,
                         {
                           <out>,
                           <query>,
                           <sort>,
                           <limit>,
                           <keytemp>,
                           <finalize>,
                           <scope>,
                           <jsMode>,
                           <verbose>
                         }
                       )
•mapreduce：指定要进行mapreduce处理的collection
•map：map函数
•reduce：reduce函数
•out：输出结果的collection的名字，不指定会默认创建一个随机名字的collection（如果使用了out选项，就不必指定keeptemp：true了，因为已经隐含在其中了）
•query：对目标集合进行过滤（query，limit，sort可以随意组合）
•sort：和limit结合的sort排序参数（也是在发往map函数前给文档排序），可以优化分组机制
•limit：发往map函数的文档数量的上限（要是没有limit，单独使用sort的用处不大）
•keytemp：是否保存临时结果，如果想在连接关闭后仍然保留这个集合，就要指定keeptemp为true，如果你用的是MongoDB的mongo客户端连接，那必须exit后才会删除。如果是脚本执行，脚本退出或调用close会自动删除结果collection
•finalize：处理结果的最后函数
•scope：向map、reduce、finalize导入外部变量。
•verbose：是否显示详细的信息

  

   MapReduce函数返回的文档结果如下:

   

{ result : <collection_name>, 

    timeMillis : <job_time>, 

    counts : { 

               input : <number of objects scanned>, 

               emit : <number of times emit was called>, 

               output : <number of items in output collection>

     } ,

     ok : <1_if_ok>,

     [, err : <errmsg_if_error>] 

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