Mongodb Mapreduce 初窺

中關(guān)村村草

Mongodb Mapreduce 初窺

聲明：本文是學(xué)習(xí)Mongodb過程中的副產(chǎn)品，因?yàn)榻佑|時(shí)間并不長，難免有理解上的偏差，希望借此文與感興趣的朋友討論切磋，呵呵。

去年年底，開始接觸并學(xué)習(xí)Mapreduce模型。因?yàn)楣ぷ魃系年P(guān)系，最近開始研究Mongodb，其中對其新特性(2010年四月)reduce模型實(shí)現(xiàn)產(chǎn)生的興趣，因?yàn)樘貏e留意了一下。當(dāng)然網(wǎng)上關(guān)于該方面的內(nèi)容并不是很多，且多為EN文，所以我想有必要將學(xué)習(xí)使用過程中的一些問題作一下記錄并加以整理，因?yàn)榫陀辛舜宋摹?br />
廢話不多說了，開始正文吧！

目前支持Mongodb的C#客戶端應(yīng)該就是Samuel Corder 開源的這個(gè)項(xiàng)目了，鏈接：http://github.com/samus/mongodb-csharp。

其中在它的源碼包中的MongoDB.Net-Tests目錄下有對TestMapReduce和TestMapReduceBuilder相應(yīng)測試用例，因?yàn)槲冶镜貨]安裝NUnit，所以接下來的內(nèi)容我是在一個(gè)新建的web項(xiàng)目中直接Copy其中的部分代碼做的測試（注：有關(guān)Mapreduce模型的內(nèi)容請查閱相關(guān)資料）。

首先我們要先加載測試數(shù)據(jù)，這里我們以DNT中的在線用戶列表的（結(jié)構(gòu)）作為依據(jù)，批量倒入10條記錄，代碼如下：

1. Mongo db = new Mongo("Servers=10.0.4.66:27017;ConnectTimeout=300000;ConnectionLifetime=300000;MinimumPoolSize=25;MaximumPoolSize=25;Pooled=true");
   
2. 
   
3. db.Connect();
   
4. 
   
5. Database test = db.GetDatabase("test");
   
6. IMongoCollection things = test["things"];
   
7. 
   
8. for (int i = 1; i <= 10;i++)
   
9. {
   
10. Document record = new Document();
    
11. record["_id"] = i;
    
12. record["userid"] = i;
    
13. record["ip"] = "10.0.7." + i;
    
14. record["username"] = "用戶" + i;
    
15. record["nickname"] = "用戶" + i;
    
16. record["password"] = "";
    
17. record["groupid"] = i;//下面將就該字段使用MAPREDUCE方式進(jìn)行分組統(tǒng)計(jì)
    
18. record["olimg"] = "";
    
19. record["adminid"] = 0;
    
20. record["invisible"] = 0;
    
21. record["action"] = 0;
    
22. record["lastactivity"] = 1;
    
23. record["lastposttime"] = DateTime.Now.ToString();
    
24. record["lastpostpmtime"] = DateTime.Now.ToString();
    
25. record["lastsearchtime"] = DateTime.Now.ToString();
    
26. record["lastupdatetime"] = "1212313221231231213321";
    
27. record["forumid"] = 0;
    
28. record["forumname"] = "";
    
29. record["titleid"] = 0;
    
30. record["title"] = "";
    
31. record["verifycode"] = "";
    
32. record["newpms"] = 0;
    
33. record["newnotices"] = 0;
    
34. things.Insert(record);
    
35. 
    
36. }

復(fù)制代碼

db.Disconnect();假定目前我們有這樣一個(gè)需求，就是找出該表中用戶組（groupid）字段為5的用戶數(shù)，當(dāng)然這里我們不會(huì)使用普通的查詢方法，而是使用MAPREDUCE方式，其工作過程分為兩個(gè)階段：map階段和reduce階段。每個(gè)階段都有鍵/值對作為輸入和輸出，并且它們的類型可由程序員選擇。下面是其實(shí)現(xiàn)方式：

首先是map方法：

string mapfunction = "function() {  if(this.groupid==5) {emit({groupid : 5}, 1);} }";然后是reduce方法：

string reducefunction = "function(key, current ){" +
   "var count = 0;" +
   "for(var i in current) {" +
       "count+=current;" +
   "}" +
   "return count;" +
"};";最后我們使用下面代碼實(shí)現(xiàn)對上面MAP,REDUCE的相應(yīng)代碼綁定和MapReduce類的聲明：

MapReduce mr = mrcol.MapReduce();
mr.Map = new Code(mapfunction);
mr.Reduce = new Code(reducefunction4);
mr.Execute();
foreach (Document doc in mr.Documents)
{
int groupCount = Convert.ToInt32(doc["value"]);
}

mr.Dispose();運(yùn)行上面代碼，顯示結(jié)果如下:

當(dāng)前上面監(jiān)視窗口中的"id:"{"groupid":5},即是mapfunction中的定義，當(dāng)然如果要統(tǒng)計(jì)所有用戶組（10個(gè)用戶組）中各自的用戶數(shù)，只把將mapfunction改寫成：

string mapfunction = "function() { emit(this.groupid, 1); }";這樣，它就會(huì)按當(dāng)前用戶所屬的groupid來作為鍵（確保不重復(fù)），凡是同一組的用戶就作為輸出進(jìn)行發(fā)送（emit），emit可以理解為調(diào)用reduce方法，這里參數(shù)為1[即累加1操作]）。

目前我在網(wǎng)上打到mongodb示例基本上都是圍繞分組統(tǒng)計(jì)功能展開的。
當(dāng)然就其傳參和返回值都可以使用類似元組的方式，記得上面的“emit({groupid : 5}, 1)”代碼嗎？返回值這里也可以使用下面的方式：

1. string reducefunction = "function(key, current ){" +
   
2.    "var count = 0;" +
   
3.    "for(var i in current) {" +
   
4.        "count+=current[i];" +
   
5.    "}" +
   
6.    "return { groupcount : count };" +  //

復(fù)制代碼

注意這里的返回方式
"};";返回類型變了，取值的方式也要發(fā)生變成：

int groupCount = int.Parse(((Document)doc["value"])["groupcount"].ToString());當(dāng)然，上面的MapReduce 類的聲明使用方式過于拘謹(jǐn)，下面使用鏈?zhǔn)秸{(diào)用的方式：

1. using (MapReduceBuilder mrb = mrcol.MapReduceBuilder().Map(mapfunction).Reduce(reducefunction))
   
2. {
   
3. using (MapReduce mr = mrb.Execute())
   
4. {
   
5. foreach (Document doc in mr.Documents)
   
6. {
   
7. int groupCount = int.Parse(((Document)doc["value"])["groupcount"].ToString());
   
8. }
   
9. }

復(fù)制代碼

}返回的結(jié)果與之前的一樣，呵呵。

另外，mongodb還支持更加復(fù)雜的數(shù)據(jù)結(jié)構(gòu)，比如官司方給的下面這個(gè)數(shù)據(jù)結(jié)構(gòu)示例：

mrcol.Insert(new Document().Append("_id", 1).Append("tags", new String[]{"dog", "cat"}));
mrcol.Insert(new Document().Append("_id", 2).Append("tags", new String[]{"dog"}));
mrcol.Insert(new Document().Append("_id", 3).Append("tags", new String[]{"mouse", "cat", "dog"}));
mrcol.Insert(new Document().Append("_id", 4).Append("tags", new String[]{}));可以看出tags字段（這里暫且這么說，呵呵），就是一個(gè)字符串?dāng)?shù)組，而下面的mapreduce方法將會(huì)統(tǒng)計(jì)里面單詞dog,cat,mouse的出現(xiàn)次數(shù)：

1. string mapfunction = "function(){\n" +
   
2.    "this.tags.forEach(\n" +
   
3.        "function(z){\n" +
   
4.            "emit( z , { count : 1 } );\n" +
   
5.        "});\n" +
   
6. "};";
   
7. string reducefunction = "function( key , values ){\n" +
   
8.     "var total = 0;\n" +
   
9.     "for ( var i=0; i
   
10.         "total += values[i].count;\n" +
    
11.     "return { count : total };\n" +

復(fù)制代碼

"};";對于如何對（含）日期型數(shù)據(jù)的鍵進(jìn)行分組統(tǒng)計(jì)，下面的這個(gè)鏈接中有詳細(xì)說明（統(tǒng)計(jì)每天用戶的訪問量）：

Counting Unique Items with Map-Reduce

下面這個(gè)鏈接就是官方給出示例的文檔鏈接頁面，其中包括更加復(fù)雜的mapreduce示例：


http://www.mongodb.org/display/DOCS/MapReduce

當(dāng)然目前對于Mapreduce模式，Mongodb使用一個(gè)單獨(dú)的進(jìn)程來跑的，這主要是因?yàn)镴avaScript 引擎的限制。目前開發(fā)團(tuán)隊(duì)正在設(shè)計(jì)解決這一問題。原文：

As of right now, MapReduce jobs on a single mongod process are single threaded. This is due to a design limitation in current JavaScript engines. We are looking into alternatives to solve this issue, but for now if you want to parallelize your MapReduce jobs, you will need to either use sharding or do the aggregation client-side in your code.

另外就是到現(xiàn)在對于MONGODB那一端是如果把輸入數(shù)據(jù)劃分成等長的小數(shù)據(jù)發(fā)送到MapReduce（Hadoop把這一操作稱為input split，即輸入切片)，因?yàn)檫@一點(diǎn)對于并發(fā)運(yùn)行的作業(yè)進(jìn)行負(fù)載平衡很重要，而在 Hadoop中一個(gè)理想的切片大小往往是一個(gè)HDFS塊的大小，默認(rèn)是64 MB（Hadoop權(quán)威指南(中文版)）。

除了上面所提到了，在MONGODB的mapreduce模型中，還支持map輸出的臨時(shí)結(jié)果集的持久化，而這一特色還在文檔中專門作了如下說明：

Note on Permanent Collections

Even when a permanent collection name is specified, a temporary collection name will be used during processing. At map/reduce completion, the temporary collection will be renamed to the permanent name atomically. Thus, one can perform a map/reduce job periodically with the same target collection name without worrying about a temporary state of incomplete data. This is very useful when generating statistical output collections on a regular basis.

而如果想要持久化該臨時(shí)集合，只要將mapreduce實(shí)例的Keptemp屬性設(shè)為true,同時(shí)使用Out屬性（方法）指定輸出的集合名稱即可。

當(dāng)然就目前我測試時(shí)結(jié)果來看，在單臺(tái)機(jī)器上做這種模型測試就效率上是得不嘗失的（執(zhí)行周期太長），特別是數(shù)據(jù)量特別大（比如3000w以上），所以應(yīng)用（或運(yùn)行）場景的選擇很重要。
上面所說的示例比較簡單，都是在單一reduce任務(wù)中的執(zhí)行場景，如下圖：



而實(shí)際的生產(chǎn)環(huán)境要比上圖復(fù)雜許多，比如多reduce任務(wù)情況，在Hadoop中，如果運(yùn)行多個(gè)reduce任務(wù)，map任務(wù)會(huì)對其輸出進(jìn)行分區(qū)，為每個(gè)reduce任務(wù)創(chuàng)建一個(gè)分區(qū)(partition)。每個(gè)分區(qū)包含許多鍵(及其關(guān)聯(lián)的值)，但每個(gè)鍵的記錄都在同一個(gè)分區(qū)中。分區(qū)可以通過用戶定義的partitioner來控制。如下圖:



鑒于目前網(wǎng)上mongodb相關(guān)文檔內(nèi)容并不多，所以這里暫不多做討論了。

好了，今天的內(nèi)容就先到這里了，感興趣的朋友可以在回復(fù)中討論

清風(fēng)鳥兒

謝謝分享

lastfile

謝謝分享，果然收藏