MongoDB MapReduce（转）

MapReduce
MapReduce是一种计算模型，简单的说就是将大批量的工作（数据）分解（MAP）执行，然后再将结果合并成最终结果（REDUCE）。这样做的好处是可以在任务被分解后，可以通过大量机器进行并行计算，减少整个操作的时间。
对科班出生的程序员来说，最好的例子莫过于归并排序的例子，没错，归并排序流程就可以看作是一个MapReduce，只是我们在学校写过的归并排序程序可能还没有涉及到并行计算罢了。
上面是MapReduce的理论部分，下面说实际的应用，下面以MongoDB MapReduce为例说明。
下面是MongoDB官方的一个例子：

Java代码



> 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

> m = function(){

... this.tags.forEach(

... function(z){

... emit( z , { count : 1 } );

... }

... );

...};

> // reduce function

> r = function( key , values ){

... var total = 0;

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

... total += values[i].count;

... return { count : total };

...};

> res = db.things.mapReduce(m, r, { out : "myoutput" } );

> res

{

"result" : "myoutput",

"timeMillis" : 12,

"counts" : {

"input" : 4,

"emit" : 6,

"output" : 3

},

"ok" : 1,

}

> db.myoutput.find()

{"_id" : "cat" , "value" : {"count" : 3}}

{"_id" : "dog" , "value" : {"count" : 2}}

{"_id" : "mouse" , "value" : {"count" : 1}}

> db.myoutput.drop()

例子很简单，计算一个标签系统中每个标签出现的次数。
这里面，除了emit函数之外，所有都是标准的js语法，当然你也可以使用你所知道的所有标准js函数。而这个emit函数是非常重要的，他的作用是将一条数据放入数据分组集合，这个分组是以emit的第一个参数为key的。你可以这样理解，当你在所有需要计算的行执行完了map函数，你就得到了一组key-values对。基本key是emit中的key，values是每次emit函数的第二个参数组成的集合。
现在我们的任务就是将这一个key-values变在key-value，也就是把这一个集合变成一个单一的值。这个操作就是Reduce。
好像这里和我们前面的理论是完全一样的，其实不然。当我们的key-values中的values集合过大，会被再切分成很多个小的key-values块，然后分别执行Reduce函数，再将多个块的结果组合成一个新的集合，作为Reduce函数的第二个参数，继续Reducer操作。可以预见，如果我们初始的values非常大，可能还会对第一次分块计算后组成的集合再次Reduce。这就类似于多阶的归并排序了。具体会有多少重，就看数据量了。
上面这一内部机制，我们不必非常了解，但我们必须了解这一机制会要求我们遵守的原则，那就是当我们书写Map函数时，emit的第二个参数形式是我们的Reduce函数的第二个参数，而Reduce函数的返回值，可能会作为新的输入参数再次执行Reduce操作，所以Reduce函数的返回值也需要和Reduce函数的第二个参数结构一致。
作为结束，下面照本宣科说一下MongoDB MapReduce调用参数和返回结果。
参数表如下：

Java代码



db.runCommand(

{ mapreduce :

,

map :

,

reduce :

[, query : ]

[, sort : ]

[, limit : ]

[, out : ]

[, keeptemp:

]

[, finalize : ]

[, scope : ]

[, verbose : true]

}

);

mapreduce：指定要进行mapreduce处理的collection

map：map函数

reduce：reduce函数

query：一个筛选条件，只有满足条件的行才会加入mapreduce集合，而这个筛选过程是先于整个mapreduce流程而执行的

sort：和query结合的sort排序参数，这是唯一可以优化分组机制的地方

limit：同上

out：结果输出的collection的名字，不指定会默认创建一个随机名字的collection

keytemp：true或false，表明结果输出到的collection是否是临时的，如果为true，则会在客户端连接中断后自动删除，如果你用的是MongoDB的mongo客户端连接，那必须exit后才会删除。如果是脚本执行，脚本退出或调用close会自动删除结果collection

finalize：和map，reduce一样是一个函数，它可以在reduce得出一个结果后再对key和value进行一次计算并返回一个最终结果

scope：设置参数值，在这里设置的值在map，reduce，finalize函数中可见

verbose：在执行过程中打印调试信息。

返回结果结构如下：

{ result :

,

counts : {

input : ,

emit : ,

output :

} ,

timeMillis : ,

ok : <1_if_ok>,

[, err : ]

}

result：储存结果的collection的名字

input：满足条件的数据行数

emit：emit调用次数，也就是所有集合中的数据总量

ouput：返回结果条数

timeMillis：执行时间，毫秒为单位

ok：是否成功，成功为1

err：如果失败，这里可以有失败原因，不过从经验上来看，原因比较模糊，作用不大

再加一个例子：

Java代码



//插入一组数据

for (var i = 0; i < 100; i++) {var u = { name : "user" + i, age : i % 40 + 1, sex : i % 2,atima:new Date().getTime() };db.users.insert(u);}

//map function

m=function(){emit(this.name,{total:this.age,lasttime:this.atima})}

//reduce function，key对应的maxtime最新的时间

r=function(key,values){var sum=0;var maxtime=0;values.forEach(function(v){sum+=v.total;if(v.lasttime>maxtime)maxtime=v.lasttime;}); return {total:sum,lasttime:maxtime};}

//运行，结果输入到result集合里

db.runCommand({mapreduce:'users',map:m,reduce:r,out:'result'});

db.result.find();