1、MapReduce 分布式計算平臺編程示例文檔名稱MapReduce 分布式計算平臺編程示例項目名稱Hadoop分布式系統研究，改進，應用作者李相娜，王守彥，馬如悅，朱冠胤文檔提交日期2008-4-3版本Versio n 1.01.MapReduce 介紹 21.1 編程模式 22.2簡單例子 22用戶自定義接口 42.1 map 函數 42.2 Reduce 函數 42.3輸入和輸出格式 52.4 partitio ner函數 52.5 Combi ner 函數 53 Hadoop MapReduce平臺使用 63.1 streami ng 介紹 63.2 C 語言Map-Reduce 程序

2、示例 73.2.1計算任務 73.2.2 M apper 算法設計 73.2.3 Reducer 算法設計 83.2.4 作業提交命令 103.3 shell Map-Reduce 程序示例 103.3.1 計算任務 103.3.2 map 實現 103.3.3 reduce 實現 113.3.4 作業提交命令 124 技巧 134.1 順序保證 134.2 本地執行 134.3 狀態信息 135 實際經驗 145.1 spider rubbish-mine 全庫挖掘項目 . 145.2 Rank 組 page 共現信息計算 155.3 PS 日志分析計算 165.4 用戶訪問信息展現 176

3、 參考資料 1718附錄1. MapReduce 介紹MapReduce是一個用于處理海量數據的分布式計算框架。這個框架解決了諸如數據分布、工作調度、容錯、機器間通信等復雜問題，可以使沒有并發處理或者分布式系 統經驗的工程師，也能很輕松地寫出簡單的、能夠應用于成百上千臺機器上處理大規模數據的并發程序。1.1編程模式MapReduce 基于分而治之的思想，將計算任務抽象成Map和Reduce兩個計算過程，其實可以簡單理解為 分散運算一合并結果”的過程。一個MapReduce 任務 首先會把輸入數據分割成不相關的若干鍵/值對（key/value ）集，這些鍵/值對會由多個map任務來并行地處理（一

4、個集合劃分對應一個 map任務）。MapReduce 會對map 的輸出（一組中間鍵值對）進行排序，并將同屬于一個鍵（key ）的值（value ）組合在一起（key/list of values ）作為reduce 任務的輸入，由reduce 任務計算出最終 結果并輸出。基本流程如下：（in put） -map - -comb ine- -red通常計算任務的輸入和輸出都是存放在文件里的，并且這些文件被存放在MapReduce之下的分布式文件系統（DFS ）中，系統會盡量使存儲結點同時成為相應 的計算節點，減少數據在網絡中流動，最大程度地節省帶寬消耗。2.2簡單例子下面給出一個詞頻統計的簡單

5、例子來演示一下 Map-Reduce 的過程：輸入兩篇文檔：doc1: MapReduce is a program ming modeldoc2: MapReduce is easy to useMap 任務負責統計每篇文檔中出現的單詞， Reduce 任務負責對每個單詞出現的總 次數進行統計。如果將每一篇文檔作為一個輸入劃分，計算結果只輸出到一個文件中； MapReduce 將啟動 2 個 map 任務和 1 個 reduce 任務進行處理。 數據轉換過程如下：Map 輸出：Map 1: ( MapReduce, 1), (is, 1), (a, 1), (programming, 1),

6、 (model, 1).Map 2: (MapReduce, 1), (is, 1), (easy, 1), (to, 1), (use, 1).Reduce 輸入：(MapReduce, 1,1), (is, 1,1), (a, 1), (programming, 1), (model,1), (easy, 1), (to, 1), (use, 1).Reduce 輸出：(MapReduce, 2), (is, 2), (a, 1), (programming, 1), (model, 1), (easy, 1), (to, 1), (use, 1).還有一些有趣的例子，也可以簡單地通過 M

7、apReduce 計算模型來展示：分布式 Grep ： 如果 map 函數檢查輸入行，滿足條件的時候， map 函數就把本 行輸出。 reduce 函數就是一個直通函數，簡單的把中間數據輸出就可以了。URL 訪問頻率統計： map 函數處理 webpag 請求和應答( URL ， 1 )的 log 。 Reduce 函數把所有相同的 URL 的值合并，并且輸出一個成對的( URL ，總個數)。逆向 Web-Link 圖： map 函數輸出所有包含指向 target URL 的 source 網 頁，用( target,source )這樣的結構對輸出。 Reduce 函數局和所有關聯相同 ta

8、rget URL 的 source 列表，并且輸出一個 (target,list(source) 這樣的結構。主機關鍵向量指標( Term-Vector per Hosts) : 關鍵詞向量指標簡言之就 是在一個文檔或者一組文檔中的重點次出現的頻率，用 (word,frequency) 表達。 map 函數計算每一個輸入文檔(主機名字是從文檔的 URL 取出的)的關鍵詞向量，然后輸 出( hostname, 關鍵詞向量 (Term-Vector) )。 reduce 函數處理所有相同 host 的文 檔關鍵詞向量。去掉不常用的關鍵詞，并且輸出最終的 (hostname ，關鍵詞向量 ) 對。逆

9、序索引 ： map 函數分析每一個文檔， 并且產生一個序列 ( word ，documentID ) 組。 reduce 函數處理指定 word 的所有的序列組，并且對相關的 document ID 進行 排序，輸出一個 (word,list(document ID) 組。所有的輸出組，組成一個簡單的逆序 索引。通過這種方法可以很容易保持關鍵詞在文檔庫中的位置。分布式排序： map 函數從每條記錄中抽取關鍵字，并且產生 (key,record) 對。 reduce 函數原樣輸出所有的關鍵字對。可見，使用 MapReduce 框架， RD 可以只關心根據怎樣將問題進行分解，哪些操 作是 Map

10、操作，哪些操作是 Reduce 操作，很大程度上簡化了整個編程模型。2 用戶自定義接口2.1 map 函數需要由用戶提供，用于處理輸入的鍵值對，并且產生一組中間的( intermediate ) 鍵值對。 MapReduce 把所有具有相同中間 key 的中間 value 聚合在一起，然后把它 們提供給 reduce 函數 .2.2 Reduce 函數同樣也需要由用戶提供，它處理中間鍵值對、以及這個中間鍵值相關的值集合，合 并這些值，最后形成一個相對較小的值集合。通常一個單次 Reduce 執行會產生 0 個 或者 1 個輸出值。提供給 Reduce 函數的中間值是通過一個迭代器( itera

11、tor )來實現 的，這就讓我們可以處理超過內存容量的值列表。2.3 輸入和輸出格式用戶可以定義輸入數據和輸出結果的格式。輸入格式( InputFormat )定義了輸 入數據的劃分方法，并實現輸入鍵值對到相應的 map 任務中。輸出格式 ( OutputFormat)定義了 reduce 輸出數據的格式和位置。2.4 partitioner 函數用戶可以提供一個 partitioner 函數，用來根據 Reduce 任務數對 map 輸出的中 間結果進行劃分。一般默認使用 Hash 劃分的方法(例如 hash(key)mod R )，就可 以得到分散均勻的分區。不過，在某些情況下，對 key

12、 用其它的函數進行分區可能更 有用。比如，某些情況下 key 是 URL ，那么我們希望所有對單個host 的入口 URL 都保存在相同的輸出文件。為了支持類似的情況，用戶可以提供一個特定的分區函數，比 如使用 hash(hostname(urlkey)mod R 作為分區函數，讓指向同一個 hostname 的 URL 分配到相同的輸出文件中。2.5 Combiner 函數用戶可以提供一個 Combiner 函數，先將 map 輸出的中間結果在本地進行合并， 然后再通過網絡發送給 reduce 任務。例如：在單詞統計的計算中， Combiner 函數可 以將 map 的輸出- Map out

13、put: (“hi” , 1), (“Owen ” , 1), (“bye ” ,1), (“Owen ” ,1)合并為Combiner output: (“Owen ” , 2), (“bye ” , 1), (“hi ” , 1)作為最終的 map 任務輸出Combiner 函數在每一個 map任務的機器上執行。通常這個combiner 函數的代 碼和reduce的代碼實現上都是一樣的。 reduce函數和combiner 函數唯一的不同就 是MapReduce 對于這兩個函數的輸出處理上不同。對于reduce函數的輸出是直接寫 到最終的輸出文件。 對于combiner函數來說，輸出是寫到

14、中間文件， 并且會被發送到 reduce 任務中去。3 Hadoop MapReduce 平臺使用3.1 streami ng 介紹Hadoop MapReduce 是對MapReduce 框架的一個java 開源實現，并且可以運 行非java語言編寫的用戶程序。它提供了一個非常有用的工具-streaming ，可以用STDIN （標準輸入）和STDOUT （標準輸出）與用戶編寫的 Map和Reduce進行數據的 交換。任何能夠使用 STDIN和STDOUT的編程語言都可以用來編寫 MapReduce 程 序，比如我們用 Python 的sys.stdin 和sys.stdout ，或者是C中

15、的stdin 和stdout。stream ing 的使用方法：Usage: $HADOOP_HOME/bi n/hadoop -co nfig dir jar $HADOOP_HOME/hadoop-streami ng.jar optio ns常用的命令選項：-in put指定輸入文件的存放路徑-output 指定結果文件的輸出位置-mapper 指定map可執行程序、或實現類名稱-reducer 指定reduce可執行程序、或實現類名稱例如:$HADOOP_HOME/b in/hadoop jar $HADOOP_HOME/hadoop-streami ng.jar -in put /u

16、ser/me/samples/cachefile/i nput.txt-mapper xargs cat-reducer cat -output /user/me/samples/cachefile/out -jobc onf mapred.map.tasks=1 -jobc onf mapred.reduce.tasks=1 -jobc onf mapred.job .n ame=Experime nt3.2 C語言Map-Reduce 程序示例3.2.1計算任務計算Apache日志中各個Refer站點出現的次數，按降序排列輸出。日志格式如下：2214220.234 - - 23/Mar/2

17、007:00:00:00 +0800 GET / HTTP/1.1 20014 mod_gzip: 0pct. 其站點名為 3.2.2 M apper 算法設計mapper讀取標準輸入，將輸入保存在局部臨時變量m中，從 m字符串中取”http:/后只包含由0-9, a-z, A-Z和.號以及-號和_號字符組成的字符串，本程序 認為上述子串為合法的url。將url部分取出來以后存儲到glib提供的哈希表中，待輸入完成以后，通過遍歷該哈希表，輸出歸并完成以后的url部分以及該url總共出現的頻度。mapper 的輸入:84 - - 28/Mar/2007:00:00:00

18、+0800 /i?z=0&cl=2&ct=201326592&sn=&lm=-1&cm=1&sc=0&bu=&rn=21&tn= baiduimage&word=%B3%C2%BB%DB%C1%D5&pn=162 HTTP/1.1 40562 mod_gzip: &sc=0&bu=&rn=21&tn=baiduimage&word=%B3%C2%BB%DB%C1%D5& pn=108 BAIDUID=D2B56887CA8714B51143E38F04880822; iCast_Rotator_1_2=1175011181843; icast_44326_682=1GET2000pct.以上為

19、一條合法的輸入記錄， mapper 將以 ”http:/后的 i 為合法 url 為開始， 到/ 結束。因此以上記錄的 url 部分為 ,將該條記錄增加到哈希表中。Mapper 輸出：每行兩列，列之間用tab 分開。補充說明：Mapper 完成標準輸入部分的歸并操作，但不對記錄排序。HttP:/ 不做處理， 對于url 部分目前只匹配 http:/ 打頭的，對含有大寫字母如https:/ 不支持。 url 部分區分大小寫。 如 和 WWW.BAIDU.COM將作為不同的 url 統計。3.2.3 Reducer 算法設計Reducer 讀取標準輸入，將輸入的字段歸并，然后輸出和輸入類似的結果。

20、Reducer以分開每行輸入的兩列，然后以第一列為關鍵字去和上一次輸入的關鍵字比較， 如果兩者匹配， 則將該輸入列 t 分隔符后的數值部分累加到上一次的數值結果中。 如果不匹配，則作為新的紀錄重新開始，并將上一記錄的結果標準輸出。輸出 部分每行兩列，第一列為 key ， 第二列為 value 部分，兩列以 t 分割在匹配過程中參考了內核進程調度 O（1 ）算法，通過交換指針，減少了字符串拷 貝與重置的過程，提高效率Reducer 輸入：235每行兩列，列間以t 分開。Reducer 輸出：5 / 對該條記錄進行了歸并5每行兩列，列間以t 分開字符串匹配部分說明： 考慮了兩種特殊情況，舉例如下：

21、 2 10 20第一種是 后沒有數字； 此時輸出為 0第二種是 和 10 這兩列之間為一空行（該行以回車結束），此時忽略該行。因此上述輸入的輸出為： 2 030324作業提交命令./hadoop jar ./c on trib/hadoop-0.15.1-streami ng.jar -mapper /home/hadoop/access_log_mapreduce/mapper_example -reducer /home/hadoop/access_log_mapreduce

22、/reducer_example 其中：-mapper 給出了 map 程序，-reducer 給出了 reduce 程序，-input 給出了 數據源路徑，-output 給出了數據結果存儲路徑，-jobco nf mapred.job. name給出了作業名稱。【程序見附錄一】3.3 shell Map-Reduce 程序示例3.3.1計算任務同上。3.3.2 map 實現streami ng接口會讀取每一個切割塊的日志行，然后以行開頭所在的文件偏移為key值，以整行的內容為 value值，通過管道送入腳本程序，腳本程序使用awk來統計各個站點的出現次數，最后輸出 map結果為以站點url

23、為key值，url在這個切割 塊中的出現次數為value值。比如：經過streaming 接口后，通過管道送入 map腳本程序的輸入為：109 34 - - 23/Mar/2007:00:00:00 +0800 GET / HTTP/1.1 200 14 mod_gzip: 0pct. 568 24 - - 23/Mar/2007:00:00:02 +0800 GET / HTTP/1.1200 14 mod_gzip: 0pct. 722 14 - - 23/Mar/2007:00:00:05 +0800 GET / HTTP

24、/1.1200 14 mod_gzip: 0pct. map 腳本程序的輸出是： 1 2腳本代碼為：#!/usr/bin/awk -furlstr = $13;if(index(urlstr ,http:/) = 2 )urlstr = substr(urlstr elseurlstr = substr(urlstr theindex=index(urlstr if(theindex != 0)urlstr = substr(urlstr, 9, length(urlstr)-2);, 2, length(urlstr)-2);, /);,1,theindex-1);sortarrayurls

25、tr+;END for(i in sortarray) print i, sortarrayi;3.3.3 reduce 實現值送入進行累streaming 接口會將已經聚合好的 （ reduce 前的排序和聚合） key/value reduce 腳本。 reduce 腳本首先將所有 key 相同（即為同一個 url ）的 value 加，最后輸出 key/value 對。輸出為次數 /url 。比如輸入為： 1 2 3 n 2 n 2輸出則為： 6 n 4腳本代碼為：#!/usr/bi n/awk -fBEGIN num = 0;urlstr =if (NR = 1) urlstr = $

26、1; num = $2;else if (urlstr = $1) num += $2;elseprint urlstr , num; urlstr = $1;num = $2;ENDprint urlstr , num;3.3.4作業提交命令./hadoop jar ./c on trib/hadoop-0.15.1-streami ng.jar -mapper /home/hadoop/access_log_mapreduce/mapper.awk -reducer /home/hadoop/access_log_mapreduce/reducer. awk 4技巧4.1順序保證如果確保在

27、給定的分區中，中間鍵 /值對的處理順序是根據 key增量處理的，便可 以很容易生成有序的輸出文件。這對于輸出文件格式需要支持對key的隨機存取時很有用，并且對輸出數據再進行排序也變得很容易。4.2本地執行因為實際計算是分布在系統中執行的，通常是在好幾千臺計算機上進行，并且是由master機器進行動態調度的任務；所以在把用戶程序提交給MapReduce平臺之前,可以先在本地機器上按 MapReduce操作順序將程序執行一遍，確保程序都調試通過。 可以使用以下類似命令：cat in put | mapper | sort | reducer output4.3狀態信息Hadoop Name nod

28、e （ Master ）內部有一個 HTTP服務器，并且可以輸出狀態報 告。狀態頁面提供了計算的進度報告，比如有多少任務已經完成， 有多少任務正在處理， 輸入的字節數，中間數據的字節數，輸出的字節數，處理百分比，等等。用戶可以根據 這些數據來預測計算需要大約執行多長時間，是否需要為這個計算增加額外的計算資 源。這些頁面也可以用來分析為何計算執行的會比預期的慢。此外，狀態頁面也顯示了哪些 worker 失效了，以及他們失效的時候上面運行的 map 和 reduce 任務。這些信息對于調試用戶代碼中的 bug 很有幫助。5 實際經驗5.1 spider rubbish-mine 全庫挖掘項目過程：

29、 從全庫中抽詞，獲得網頁模式，根據模式相似度去除垃圾頁面。困難：數據量巨大，全部文本數據 480TB ，壓縮后 120TB ，單機抽詞速度很慢。抽詞后結果 35TB ，需要對 35TB 數據排序，把相同模式的網頁匯聚在一起，普通機群無法完成，編 程困難比較大。平臺使用解決方法：使用50臺機器搭建機群， 使用250 個核并行抽詞并局部排序， 抽詞后對結果使用分布式 排序，送給垃圾頁面處理程序處理。項目實施困難：1 網頁格式和平臺默認格式不一致；2 實驗機群異構性很強，存儲能力不平衡；3 實驗機群計算能力不平衡；4 編碼問題；5 數據導入比較慢；6 平臺容量有限，不足以存下抽詞結果；7 存儲能力極

30、不平衡時，數據導入停滯。處理能力：1 使用不壓縮方式， 200 個核每天處理 48TB 不壓縮數據；2 使用壓縮方式， 250 個核每天處理 15TB 壓縮數據，數據壓縮比為 4:1 。 遷移成本：完成程序從單機遷移到平臺上并測試耗時一星期， 由于平臺實現了分布式排序，大大簡 化了程序編寫。收益：原計劃一個月計算完成的項目遷移到 250 個核的機群上，經過多次改進和調整，目前只需 8-9 天。5.2 Rank 組 page 共現信息計算過程： 對網頁進行抽詞，對抽詞結果兩兩組合，統計所有網頁中按各種組合出現的總次數。困難：數據量接近 1TB ，抽詞組成詞對后數據量放大 5 倍，抽詞組合速度很慢

31、。自己構建機群 處理，程序實現周期比較長，節點失效后不能自動處理，無法實現計算資源和存儲計算 的平衡，節點數超過一定數量后難于維護；中間結果巨大，如果使用少量機器，網絡壓 力很大。平臺使用解決方法：使用50 臺機器并行進行多輪抽詞，多抽詞結果做兩次 reduce 計算，完成最終結果。項目實施中的困難：1 中間結果巨大，但是平臺網絡異構比較嚴重，為了保障其他業務不受影響，只能 把任務切割的盡量小；2 在本次計算中，根據資源情況，參數配置并沒有達到最優, 未能發揮平臺最佳性能；3 編碼問題。遷移成本：遷移到分布式平臺的成本適中， 提取出核心代碼就能夠在分布式平臺上使用，源程序的 大部分代碼為數據分

32、發功能， 使用分布式平臺大大減少了代碼編寫量。 大概使用一天時 間完成代碼遷移和單元測試。收益：NLP搭建了 7臺機器的小分布式系統，該系統使用 2周時間處理了 3000W網頁，計劃至 少處理 1億以上網頁，與預期相差太多，只好使用分布式計算平臺，使用分布式平臺的 最大收益是計算時間的大量較少，本次計算了 2億網頁使用了 18 小時，能夠滿足時間要 求。5.3 PS 日志分析計算需求： 競爭對手流量來源分析，用戶主動輸入 url 分析，大網站用戶訪問路徑挖掘輸入數據量 在200GB-400GB 之間，屬于分布式平臺上的典型應用。遷移成本： 基本沒有遷移成本，平臺可以實現分布式排序，簡化了程序編

33、寫。收益：之前使用單機進行數據分析，一個分析需要耗時 2小時左右，使用分布式平臺耗時在 51個月以上的日志數據，用分鐘以內。使用分布式平臺提供的大容量存儲能力可以保存來做深度分析。5.4用戶訪問信息展現需求：把用戶一天的訪問信息導入數據平臺，pm可以根據關鍵字迅速找到訪問數據，用于分析用戶行為。困難：數據量巨大，單機平臺（數據庫）效率低，之前使用RD自己辦些的索引系統實現，實現周期比較長，可擴展性比較差。平臺使用解決辦法：使用在分布式存儲之上構建的分布式索引系統，相當于分布式數據庫，在數據倒入時， 自動建索引，查詢時也使用分布式查詢。遷移成本：編寫程序把數據倒入即可，不需要考慮索引和效率問題。

34、項目實施困難：目前的分布式索引系統還不太穩定，只適用于線下系統。收益：降低了系統實現難度和維護成本，增加了可擴展性。6參考資料1. Google MapReduce paper2. Hadoop Map-Reduce Tutorial3. Hadoop Stream ing附錄mapper_example.c/* On platform 32:gcc -I/usr/include/glib-2.0 -I/usr/lib/glib-2.0/include-lglib-2.0 -o mappermapper_example.c -O3* On platform 64: gcc -I/usr/inc

35、lude/glib-2.0 -I/usr/lib64/glib-2.0/include -lglib-2.0 -o mappermapper_example.c -O3*/#include #include #include #include #define HTTP_HEADER_LEN 9#define HTTP_HEADER http:/#define MAX_LEN 8192#define MAX_KEY_LEN 256#define INIT 1#define HASH_FOUND 1#define HASH_NOT_FOUND 0#define FREE(ptr) if(ptr)!

36、=NULL) free(ptr);(ptr) = NULL;static GHashTable* g_url_hash = NULL;/* run for each hash pair, called by g_hash_table_foreach()*/inline void hash_table_iterator(gpointer key, gpointer value, gpointer user_data) printf(%st%un, key, *(long*)value);FREE(key);FREE(value);/* function: update the keys valu

37、e in hash, if key exist, update value, if key not exist, insert intohash.* return: 1 if key exist in hash, 0 for not exist.*/longinline int hash_table_update(GHashTable *hash, const char *key, value)char *pkey= NULL;long *pvalue = NULL;int length = 0;int res = -1;pvalue = g_hash_table_lookup(hash, k

38、ey);if(pvalue = NULL) res = HASH_NOT_FOUND; length = strlen(key);pkey = (char*)malloc(sizeof(char)*(length+1); memset(void*)pkey, 0, length+1); memcpy(pkey, key, length);pvalue = (long*)malloc(sizeof(long);*pvalue = value; g_hash_table_insert(hash, pkey, pvalue); else *(long*)pvalue += value; res =

39、HASH_FOUND;return res;/* compute next value for sting s.*/void getnext(const char *s, int next)int i,j; i=0;j=-1;next0=-1; while(si)start from 0,next, charif(j=-1|si=sj) +i;+j;nexti=j; else j=nextj;/* return the offset in string m, if string s was founded in m. -1 means not found.*/inline int kmp(co

40、nst char *m, const char *s, const int next)int i,j; i=0;j=0; while(mi) if(j=-1|mi=sj) +i;+j; if(sj=0) return (i-j); else j=nextj;return -1;/* Function: using kmp algorithm to math string s in string m, store the matched string in key.* Return: NULL for not matched, or key pointer.*/inline char* getk

41、ey(const char *m, const char *s, const int *key)int urlStart = 0, urlEnd = 0, kmpres = 0;kmpres = kmp(m, s, next); if(kmpres = -1) return NULL;urlStart= urlEnd =kmpres + strlen(s);while(murlEnd&(murlEnd=a&murlEnd=0&murlEnd=A&murlEnd=Z)|(murlEnd=-)|(murlEnd=_) +urlEnd;if(urlEnd!=urlStart) memset(void

42、*)key, 0, sizeof(key); memcpy(key, &murlStart, urlEnd-urlStart); return key;return NULL;int main(int argc, char *argv)char mMAX_LEN,sHTTP_HEADER_LEN;char keystrMAX_KEY_LEN;int nextHTTP_HEADER_LEN;memset(void*)m, 0, sizeof(m); memset(void*)keystr, 0, sizeof(keystr);/ init s string (kmp algorithm) mem

43、set(void*)s, 0, sizeof(s); snprintf(s, sizeof(s), %s, HTTP_HEADER);/ compute next value (kmp algorithm) memset(void*)next, 0, sizeof(next); getnext(s,next);/init g_url_hashg_url_hash = g_hash_table_new(g_str_hash, g_str_equal);while(fgets(m, MAX_LEN, stdin) != NULL)hashif(getkey(m, s, next, keystr) != NULL) hash_table_update(g_url_hash, keystr , 1);/add pair(keystr , 1) int