MongoDB MapReduce
MapReduce是一種計(jì)算模型，簡(jiǎn)單的說(shuō)就是將大批量的工作（數(shù)據(jù)）分解（MAP）執(zhí)行，然后再將結(jié)果合并成最終結(jié)果（REDUCE）。這樣做的好處是可以在任務(wù)被分解后，可以通過(guò)大量機(jī)器進(jìn)行并行計(jì)算，減少整個(gè)操作的時(shí)間。
上面是MapReduce的理論部分，下面說(shuō)實(shí)際的應(yīng)用，下面以MongoDB MapReduce為例說(shuō)明。
下面是MongoDB官方的一個(gè)例子：
> 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( z , { count : 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 };
...};
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" : "mouse", "value" : { "count" : 1 } }
例子很簡(jiǎn)單，計(jì)算一個(gè)標(biāo)簽系統(tǒng)中每個(gè)標(biāo)簽出現(xiàn)的次數(shù)。
這里面，除了emit函數(shù)之外，所有都是標(biāo)準(zhǔn)的js語(yǔ)法,這個(gè)emit函數(shù)是非常重要的，可以這樣理解，當(dāng)所有需要計(jì)算的文檔（因?yàn)樵趍apReduce時(shí)，可以對(duì)文檔進(jìn)行過(guò)濾，接下來(lái)會(huì)講到）執(zhí)行完了map函數(shù)，map函數(shù)會(huì)返回key_values對(duì)，key即是emit中的第一個(gè)參數(shù)key，values是對(duì)應(yīng)同一key的emit的n個(gè)第二個(gè)參數(shù)組成的數(shù)組。這個(gè)key_values會(huì)作為參數(shù)傳遞給reduce，分別作為第1.2個(gè)參數(shù)。
reduce函數(shù)的任務(wù)就是將key-values變成key-value，也就是把values數(shù)組變成一個(gè)單一的值value。當(dāng)key-values中的values數(shù)組過(guò)大時(shí)，會(huì)被再切分成很多個(gè)小的key-values塊，然后分別執(zhí)行Reduce函數(shù)，再將多個(gè)塊的結(jié)果組合成一個(gè)新的數(shù)組，作為Reduce函數(shù)的第二個(gè)參數(shù)，繼續(xù)Reducer操作。可以預(yù)見(jiàn)，如果我們初始的values非常大，可能還會(huì)對(duì)第一次分塊計(jì)算后組成的集合再次Reduce。這就類(lèi)似于多階的歸并排序了。具體會(huì)有多少重，就看數(shù)據(jù)量了。
reduce一定要能被反復(fù)調(diào)用，不論是映射環(huán)節(jié)還是前一個(gè)簡(jiǎn)化環(huán)節(jié)。所以reduce返回的文檔必須能作為reduce的第二個(gè)參數(shù)的一個(gè)元素。
（當(dāng)書(shū)寫(xiě)Map函數(shù)時(shí)，emit的第二個(gè)參數(shù)組成數(shù)組成了reduce函數(shù)的第二個(gè)參數(shù)，而Reduce函數(shù)的返回值，跟emit函數(shù)的第二個(gè)參數(shù)形式要一致，多個(gè)reduce函數(shù)的返回值可能會(huì)組成數(shù)組作為新的第二個(gè)輸入?yún)?shù)再次執(zhí)行Reduce操作。）
MapReduce函數(shù)的參數(shù)列表如下：
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]
}
);
或者這么寫(xiě)：
db.collection.mapReduce(
<map>,
<reduce>,
{
<out>,
<query>,
<sort>,
<limit>,
<keytemp>,
<finalize>,
<scope>,
<jsMode>,
<verbose>
}
)
1.mapreduce：指定要進(jìn)行mapreduce處理的collection
2.map：map函數(shù)
3.reduce：reduce函數(shù)
4.out：輸出結(jié)果的collection的名字，不指定會(huì)默認(rèn)創(chuàng)建一個(gè)隨機(jī)名字的collection（如果使用了out選項(xiàng)，就不必指定keeptemp：true了，因?yàn)橐呀?jīng)隱含在其中了）
5.query：一個(gè)篩選條件，只有滿(mǎn)足條件的文檔才會(huì)調(diào)用map函數(shù)。（query。limit，sort可以隨意組合）
6.sort：和limit結(jié)合的sort排序參數(shù)（也是在發(fā)往map函數(shù)前給文檔排序），可以?xún)?yōu)化分組機(jī)制
7.limit：發(fā)往map函數(shù)的文檔數(shù)量的上限（要是沒(méi)有l(wèi)imit，單獨(dú)使用sort的用處不大）
8.keytemp：true或false，表明結(jié)果輸出到的collection是否是臨時(shí)的，如果想在連接關(guān)閉后仍然保留這個(gè)集合，就要指定keeptemp為true，如果你用的是MongoDB的mongo客戶(hù)端連接，那必須exit后才會(huì)刪除。如果是腳本執(zhí)行，腳本退出或調(diào)用close會(huì)自動(dòng)刪除結(jié)果collection
9.finalize：是函數(shù)，它會(huì)在執(zhí)行完map、reduce后再對(duì)key和value進(jìn)行一次計(jì)算并返回一個(gè)最終結(jié)果，這是處理過(guò)程的最后一步，所以finalize就是一個(gè)計(jì)算平均數(shù)，剪裁數(shù)組，清除多余信息的恰當(dāng)時(shí)機(jī)
10.scope：javascript代碼中要用到的變量，在這里定義的變量在map，reduce，finalize函數(shù)中可見(jiàn)
11.verbose：用于調(diào)試的詳細(xì)輸出選項(xiàng)，如果想看MpaReduce的運(yùn)行過(guò)程，可以設(shè)置其為true。也可以print把map，reduce，finalize過(guò)程中的信息輸出到服務(wù)器日志上。
執(zhí)行MapReduce函數(shù)返回的文檔結(jié)構(gòu)如下：
{ 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>]
}
1.result：儲(chǔ)存結(jié)果的collection的名字,這是個(gè)臨時(shí)集合，MapReduce的連接關(guān)閉后自動(dòng)就被刪除了。
2.timeMillis：執(zhí)行花費(fèi)的時(shí)間，毫秒為單位
3.input：滿(mǎn)足條件被發(fā)送到map函數(shù)的文檔個(gè)數(shù)
4.emit：在map函數(shù)中emit被調(diào)用的次數(shù)，也就是所有集合中的數(shù)據(jù)總量
5.ouput：結(jié)果集合中的文檔個(gè)數(shù)（count對(duì)調(diào)試非常有幫助）
6.ok：是否成功，成功為1
7.err：如果失敗，這里可以有失敗原因，不過(guò)從經(jīng)驗(yàn)上來(lái)看，原因比較模糊，作用不大
java代碼執(zhí)行MapReduce的方法：
public void MapReduce() {
Mongo mongo = new Mongo("localhost",27017);
DB db = mongo.getDB("qimiguangdb");
DBCollection coll = db.getCollection("collection1");
String map = "function() { emit(this.name, {count:1});}";
String reduce = "function(key, values) {";
reduce=reduce+"var total = 0;";
reduce=reduce+"for(var i=0;i<values.length;i++){total += values[i].count;}";
reduce=reduce+"return {count:total};}";
String result = "resultCollection";
MapReduceOutput mapReduceOutput = coll.mapReduce(map,
reduce.toString(), result, null);
DBCollection resultColl = mapReduceOutput.getOutputCollection();
DBCursor cursor= resultColl.find();
while (cursor.hasNext()) {
System.out.println(cursor.next());
}
}
