hadoop常用算法例子

1、基本MapReduce模式計數(shù)與求和基本MapReduce模式計數(shù)與求和問題陳述:有許多文檔，每個文檔都有一些字段組成。需要計算出每個字段在所有文檔中的出現(xiàn)次數(shù)或者這些字段的其他什么統(tǒng)計值。例如，給定一個log文件，其中的每條記錄都包含一個響應(yīng)時間，需要計算出平均響應(yīng)時間。解決方案:讓我們先從簡單的例子入手。在下面的代碼片段里，Mapper每遇到指定詞就把頻次記1，Reducer一個個遍歷這些詞的集合然后把他們的頻次加和。1. class Mapper2.     method Map(docid id, doc d)3.       

2、;  for all term t in doc d do4.           Emit(term t, count 1)5.6. class Reducer7.     method Reduce(term t, counts c1, c2,.)8.         sum = 09.         for all count c in c1, c2,. do10. &

3、#160;           sum = sum + c11.             Emit(term t, count sum)復(fù)制代碼這種方法的缺點顯而易見，Mapper提交了太多無意義的計數(shù)。它完全可以通過先對每個文檔中的詞進行計數(shù)從而減少傳遞給Reducer的數(shù)據(jù)量:size=14.166666030883789px1. 1 class Mapper2. 2   method Map(docid id, doc

4、d)3. 3     H = new AssociativeArray4. 4     for all term t in doc d do5. 5       Ht = Ht + 16. 6       for all term t in H do7. 7         Emit(term t, count Ht)復(fù)制代碼如果要累計計數(shù)的的不只是單個文檔中的內(nèi)容，還包括了一

5、個Mapper節(jié)點處理的所有文檔，那就要用到Combiner了:size=14.166666030883789px1. 1   class Mapper2. 2      method Map(docid id, doc d)3. 3         for all term t in doc d do4. 4            Emit(term t, count 1)5. 5 6

6、. 6   class Combiner7. 7      method Combine(term t, c1, c2,.)8. 8         sum = 09. 9         for all count c in c1, c2,. do10. 10             sum = sum + c11. 11 

7、       Emit(term t, count sum)12. 12 13. 13   class Reducer14. 14      method Reduce(term t, counts c1, c2,.)15. 15         sum = 016. 16         for all count c in c1, c2,. do17.

8、17             sum = sum + c18. 18         Emit(term t, count sum)復(fù)制代碼應(yīng)用：Log 分析, 數(shù)據(jù)查詢整理歸類問題陳述:有一系列條目，每個條目都有幾個屬性，要把具有同一屬性值的條目都保存在一個文件里，或者把條目按照屬性值分組。 最典型的應(yīng)用是倒排索引。解決方案：解決方案很簡單。 在 Mapper 中以每個條目的所需屬性值作為 key，其本身作為值傳遞給 Reducer。 Reducer 取

9、得按照屬性值分組的條目，然后可以處理或者保存。如果是在構(gòu)建倒排索引，那么 每個條目相當(dāng)于一個詞而屬性值就是詞所在的文檔ID。應(yīng)用：倒排索引， ETL過濾 (文本查找)，解析和校驗問題陳述:假設(shè)有很多條記錄，需要從其中找出滿足某個條件的所有記錄，或者將每條記錄傳換成另外一種形式（轉(zhuǎn)換操作相對于各條記錄獨立，即對一條記錄的操作與其他記錄無關(guān)）。像文本解析、特定值抽取、格式轉(zhuǎn)換等都屬于后一種用例。解決方案:非常簡單，在Mapper 里逐條進行操作，輸出需要的值或轉(zhuǎn)換后的形式。應(yīng)用：日志分析，數(shù)據(jù)查詢，ETL，數(shù)據(jù)校驗分布式任務(wù)執(zhí)行問題陳述:大型計算可以分解為多個部分分別進行然后合并各個計算的結(jié)果以獲

10、得最終結(jié)果。解決方案:  將數(shù)據(jù)切分成多份作為每個 Mapper 的輸入，每個Mapper處理一份數(shù)據(jù)，執(zhí)行同樣的運算，產(chǎn)生結(jié)果，Reducer把多個Mapper的結(jié)果組合成一個。案例研究： 數(shù)字通信系統(tǒng)模擬像 WiMAX 這樣的數(shù)字通信模擬軟件通過系統(tǒng)模型來傳輸大量的隨機數(shù)據(jù)，然后計算傳輸中的錯誤幾率。 每個 Mapper 處理樣本 1/N  的數(shù)據(jù)，計算出這部分?jǐn)?shù)據(jù)的錯誤率，然后在 Reducer 里計算平均錯誤率。應(yīng)用：工程模擬，數(shù)字分析，性能測試排序問題陳述:有許多條記錄，需要按照某種規(guī)則將所有記錄排序或是按照順序來處理記錄。解決方案: 

11、;簡單排序很好辦 Mappers 將待排序的屬性值為鍵，整條記錄為值輸出。 不過實際應(yīng)用中的排序要更加巧妙一點， 這就是它之所以被稱為MapReduce 核心的原因（“核心”是說排序？因為證明Hadoop計算能力的實驗是大數(shù)據(jù)排序？還是說Hadoop的處理過程中對key排序的環(huán)節(jié)？）。在實踐中，常用組合鍵來實現(xiàn)二次排序和分組。MapReduce 最初只能夠?qū)︽I排序， 但是也有技術(shù)利用可以利用Hadoop 的特性來實現(xiàn)按值排序。想了解的話可以看這篇博客。按照BigTable的概念，使用 MapReduce來對最初數(shù)據(jù)而非中間數(shù)據(jù)排序，也即保持?jǐn)?shù)據(jù)的有序狀態(tài)更有好處，必須注意這一點。換句話說，在數(shù)

12、據(jù)插入時排序一次要比在每次查詢數(shù)據(jù)的時候排序更高效。應(yīng)用：ETL，數(shù)據(jù)分析非基本 MapReduce 模式迭代消息傳遞 (圖處理)問題陳述：假設(shè)一個實體網(wǎng)絡(luò)，實體之間存在著關(guān)系。 需要按照與它比鄰的其他實體的屬性計算出一個狀態(tài)。這個狀態(tài)可以表現(xiàn)為它和其它節(jié)點之間的距離， 存在特定屬性的鄰接點的跡象， 鄰域密度特征等等。解決方案：網(wǎng)絡(luò)存儲為系列節(jié)點的結(jié)合，每個節(jié)點包含有其所有鄰接點ID的列表。按照這個概念，MapReduce 迭代進行，每次迭代中每個節(jié)點都發(fā)消息給它的鄰接點。鄰接點根據(jù)接收到的信息更新自己的狀態(tài)。當(dāng)滿足了某些條件的時候迭代停止，如達到了最大迭代次數(shù)（網(wǎng)絡(luò)半徑）或兩次連續(xù)的迭代幾乎

13、沒有狀態(tài)改變。從技術(shù)上來看，Mapper 以每個鄰接點的ID為鍵發(fā)出信息，所有的信息都會按照接受節(jié)點分組，reducer 就能夠重算各節(jié)點的狀態(tài)然后更新那些狀態(tài)改變了的節(jié)點。下面展示了這個算法：size=14.166666030883789px1. 1 class Mapper2. 2   method Map(id n, object N)3. 3     Emit(id n, object N)4. 4     for all id m in N.OutgoingRelations do5.

14、 5       Emit(id m, message getMessage(N)6. 6 7. 7 class Reducer8. 8   method Reduce(id m, s1, s2,.)9. 9     M = null10. 10     messages = 11. 11     for all s in s1, s2,. do12. 12      

15、; if IsObject(s) then13. 13         M = s14. 14       else               / s is a message15. 15         messages.add(s)16. 16     M.State = calculateS

16、tate(messages)17. 17     Emit(id m, item M)復(fù)制代碼一個節(jié)點的狀態(tài)可以迅速的沿著網(wǎng)絡(luò)傳全網(wǎng)，那些被感染了的節(jié)點又去感染它們的鄰居，整個過程就像下面的圖示一樣：案例研究： 沿分類樹的有效性傳遞· 案例研究： 沿分類樹的有效性傳遞問題陳述：這個問題來自于真實的電子商務(wù)應(yīng)用。將各種貨物分類，這些類別可以組成一個樹形結(jié)構(gòu)，比較大的分類（像男人、女人、兒童）可以再分出小分類（像男褲或女裝），直到不能再分為止（像男式藍色牛仔褲）。這些不能再分的基層類別可以是有效（這個類別包含有貨品）或者已無效的（沒有屬于這個分類的

17、貨品）。如果一個分類至少含有一個有效的子分類那么認(rèn)為這個分類也是有效的。我們需要在已知一些基層分類有效的情況下找出分類樹上所有有效的分類。解決方案：這個問題可以用上一節(jié)提到的框架來解決。我們咋下面定義了名為 getMessage和 calculateState 的方法：size=14.166666030883789px1. 1 class N2. 2   State in True = 2, False = 1, null = 0,3. 3   initialized 1 or 2 for end-of-line categories, 0 otherw

18、ise4. 4   method getMessage(object N)5. 5     return N.State6. 6   method calculateState(state s, data d1, d2,.)7. 7     return max( d1, d2,. )復(fù)制代碼案例研究：廣度優(yōu)先搜索問題陳述：需要計算出一個圖結(jié)構(gòu)中某一個節(jié)點到其它所有節(jié)點的距離。解決方案： Source源節(jié)點給所有鄰接點發(fā)出值為0的信號，鄰接點把收到的信號再轉(zhuǎn)發(fā)給自己

19、的鄰接點，每轉(zhuǎn)發(fā)一次就對信號值加1：size=14.166666030883789px8. 1 class N9. 2   State is distance,10. 3   initialized 0 for source node, INFINITY for all other nodes11. 4   method getMessage(N)12. 5     return N.State + 113. 6   method calculateState(state s

20、, data d1, d2,.)14. 7     min( d1, d2,. )復(fù)制代碼案例研究：網(wǎng)頁排名和 Mapper 端數(shù)據(jù)聚合這個算法由Google提出，使用權(quán)威的PageRank算法，通過連接到一個網(wǎng)頁的其他網(wǎng)頁來計算網(wǎng)頁的相關(guān)性。真實算法是相當(dāng)復(fù)雜的，但是核心思想是權(quán)重可以傳播，也即通過一個節(jié)點的各聯(lián)接節(jié)點的權(quán)重的均值來計算節(jié)點自身的權(quán)重。size=14.166666030883789px15. 1 class N16. 2   State is PageRank17. 3   method getM

21、essage(object N)18. 4     return N.State / N.OutgoingRelations.size()19. 5   method calculateState(state s, data d1, d2,.)20. 6     return ( sum(d1, d2,.) )復(fù)制代碼要指出的是上面用一個數(shù)值來作為評分實際上是一種簡化，在實際情況下，我們需要在Mapper端來進行聚合計算得出這個值。下面的代碼片段展示了這個改變后的邏輯 （針對于 PageRank

22、 算法）：size=14.166666030883789px21. 1 class Mapper22. 2   method Initialize23. 3     H = new AssociativeArray24. 4   method Map(id n, object N)25. 5     p = N.PageRank  / N.OutgoingRelations.size()26. 6     Emit(

23、id n, object N)27. 7     for all id m in N.OutgoingRelations do28. 8       Hm = Hm + p29. 9   method Close30. 10     for all id n in H do31. 11       Emit(id n, value Hn)32. 12 33. 13 class Reducer34. 14&#

24、160;  method Reduce(id m, s1, s2,.)35. 15     M = null36. 16     p = 037. 17     for all s in s1, s2,. do38. 18       if IsObject(s) then39. 19         M = s40. 20     &#

25、160; else41. 21         p = p + s42. 22         M.PageRank = p43. 23         Emit(id m, item M)復(fù)制代碼應(yīng)用：圖分析，網(wǎng)頁索引值去重 （對唯一項計數(shù)）問題陳述: 記錄包含值域F和值域 G，要分別統(tǒng)計相同G值的記錄中不同的F值的數(shù)目 (相當(dāng)于按照 G分組).這個問題可以推而廣之應(yīng)用于分面搜索（某些電子商務(wù)網(wǎng)站稱之為N

26、arrow Search）44.   Record 1: F=1, G=a, b45.   Record 2: F=2, G=a, d, e46.   Record 3: F=1, G=b47.   Record 4: F=3, G=a, b48.49.   Result:50.   a -> 3 / F=1, F=2, F=351.   b -> 2 / F=1, F=352.   d -> 1 / F=253. &

27、#160; e -> 1 / F=2復(fù)制代碼解決方案 I:第一種方法是分兩個階段來解決這個問題。第一階段在Mapper中使用F和G組成一個復(fù)合值對，然后在Reducer中輸出每個值對，目的是為了保證F值的唯一性。在第二階段，再將值對按照G值來分組計算每組中的條目數(shù)。第一階段：54. 1   class Mapper55. 2     method Map(null, record value f, categories g1, g2,.)56. 3       for all c

28、ategory g in g1, g2,.57. 4         Emit(record g, f, count 1)58. 5 59. 6   class Reducer60. 7     method Reduce(record g, f, counts n1, n2, .)61. 8       Emit(record g, f, null )復(fù)制代碼第二階段：62. 1   class Mapp

29、er63. 2     method Map(record f, g, null)64. 3       Emit(value g, count 1)65. 4 66. 5   class Reducer67. 6     method Reduce(value g, counts n1, n2,.)68. 7       Emit(value g, sum( n1, n2,. ) )復(fù)制代碼解決方案 II

30、:第二種方法只需要一次MapReduce 即可實現(xiàn)，但擴展性不強。算法很簡單-Mapper 輸出值和分類，在Reducer里為每個值對應(yīng)的分類去重然后給每個所屬的分類計數(shù)加1，最后再在Reducer結(jié)束后將所有計數(shù)加和。這種方法適用于只有有限個分類，而且擁有相同F(xiàn)值的記錄不是很多的情況。例如網(wǎng)絡(luò)日志處理和用戶分類，用戶的總數(shù)很多，但是每個用戶的事件是有限的，以此分類得到的類別也是有限的。值得一提的是在這種模式下可以在數(shù)據(jù)傳輸?shù)絉educer之前使用Combiner來去除分類的重復(fù)值。69. 1 class Mapper70. 2   method Map(null, rec

31、ord value f, categories g1, g2,. )71. 3     for all category g in g1, g2,.72. 4       Emit(value f, category g)73. 5 74. 6 class Reducer75. 7   method Initialize76. 8     H = new AssociativeArray : category -> count77. 9&#

32、160;  method Reduce(value f, categories g1, g2,.)78. 10     g1', g2',. = ExcludeDuplicates( g1, g2,. )79. 11     for all category g in g1', g2',.80. 12       Hg = Hg + 181. 13   method Close82. 14   

33、  for all category g in H do83. 15       Emit(category g, count Hg)復(fù)制代碼應(yīng)用：日志分析，用戶計數(shù)互相關(guān)問題陳述：有多個各由若干項構(gòu)成的組，計算項兩兩共同出現(xiàn)于一個組中的次數(shù)。假如項數(shù)是N，那么應(yīng)該計算N*N。這種情況常見于文本分析（條目是單詞而元組是句子），市場分析（購買了此物的客戶還可能購買什么）。如果N*N小到可以容納于一臺機器的內(nèi)存，實現(xiàn)起來就比較簡單了。配對法第一種方法是在Mapper中給所有條目配對，然后在Reducer中將同一條目對的計數(shù)加和。但這種做法

34、也有缺點：· 使用 combiners 帶來的的好處有限，因為很可能所有項對都是唯一的· 不能有效利用內(nèi)存size=14.166666030883789px86. 1 class Mapper87. 2   method Map(null, items i1, i2,. )88. 3     for all item i in i1, i2,.89. 4       for all item j in i1, i2,.90. 5     

35、0;   Emit(pair i j, count 1)91. 6 92. 7 class Reducer93. 8   method Reduce(pair i j, counts c1, c2,.)94. 9     s = sum(c1, c2,.)95. 10     Emit(pairi j, count s)復(fù)制代碼Stripes Approach（條方法？不知道這個名字怎么理解）第二種方法是將數(shù)據(jù)按照pair中的第一項來分組，并維護一個關(guān)聯(lián)數(shù)組，數(shù)組中

36、存儲的是所有關(guān)聯(lián)項的計數(shù)。The second approach is to group data by the first item in pair and maintain an associative array (“stripe”) where counters for all adjacent items are accumulated. Reducer receives all stripes for leading item i, merges them, and emits the same result as in the Pairs approach.· 中間結(jié)果

37、的鍵數(shù)量相對較少，因此減少了排序消耗。· 可以有效利用 combiners。· 可在內(nèi)存中執(zhí)行，不過如果沒有正確執(zhí)行的話也會帶來問題。· 實現(xiàn)起來比較復(fù)雜。· 一般來說， “stripes” 比 “pairs” 更快size=14.166666030883789px101. 1 class Mapper102. 2   method Map(null, items i1, i2,. )103. 3     for all item i in i1, i2,.104. 4   

38、    H = new AssociativeArray : item -> counter105. 5       for all item j in i1, i2,.106. 6         Hj = Hj + 1107. 7         Emit(item i, stripe H)108. 8 109. 9 class Reducer110. 10   method

39、Reduce(item i, stripes H1, H2,.)111. 11     H = new AssociativeArray : item -> counter112. 12     H = merge-sum( H1, H2,. )113. 13     for all item j in H.keys()114. 14       Emit(pair i j, Hj)復(fù)制代碼應(yīng)用：文本分析，市場分析· 參

40、考資料：Lin J. Dyer C. Hirst G. Data Intensive Processing MapReduce用MapReduce 表達關(guān)系模式在這部分我們會討論一下怎么使用MapReduce來進行主要的關(guān)系操作。篩選（Selection）· size=14.166666030883789px1. 1 class Mapper2. 2   method Map(rowkey key, tuple t)3. 3   if t satisfies the predicate4. 4    &#

41、160;Emit(tuple t, null)復(fù)制代碼投影（Projection）投影只比篩選稍微復(fù)雜一點，在這種情況下我們可以用Reducer來消除可能的重復(fù)值。size=14.166666030883789px5. 1 class Mapper6. 2   method Map(rowkey key, tuple t)7. 3     tuple g = project(t) / extract required fields to tuple g8. 4     Emit(tuple g,

42、 null)9. 5 10. 6 class Reducer11. 7   method Reduce(tuple t, array n) / n is an array of nulls12. 8     Emit(tuple t, null)復(fù)制代碼合并（Union）兩個數(shù)據(jù)集中的所有記錄都送入Mapper，在Reducer里消重。size=14.166666030883789px13. 1 class Mapper14. 2   method Map(rowkey key, tuple t)15.

43、3     Emit(tuple t, null)16. 4 17. 5 class Reducer18. 6   method Reduce(tuple t, array n) / n is an array of one or two nulls19. 7     Emit(tuple t, null)復(fù)制代碼交集（Intersection）將兩個數(shù)據(jù)集中需要做交叉的記錄輸入Mapper，Reducer 輸出出現(xiàn)了兩次的記錄。因為每條記錄都有一個主鍵，在每個數(shù)據(jù)集中只會出現(xiàn)一次，所

44、以這樣做是可行的。size=14.166666030883789px20. 1 class Mapper21. 2   method Map(rowkey key, tuple t)22. 3     Emit(tuple t, null)23. 4 24. 5 class Reducer25. 6   method Reduce(tuple t, array n) / n is an array of one or two nulls26. 7     if n.

45、size() = 227. 8       Emit(tuple t, null)復(fù)制代碼差異（Difference）假設(shè)有兩個數(shù)據(jù)集R和S，我們要找出R與S的差異。Mapper將所有的元組做上標(biāo)記，表明他們來自于R還是S，Reducer只輸出那些存在于R中而不在S中的記錄。size=14.166666030883789px28. 1 class Mapper29. 2   method Map(rowkey key, tuple t)30. 3     Emit(tuple t, string

46、 t.SetName) / t.SetName is either 'R' or 'S'31. 4 32. 5 class Reducer33. 6   method Reduce(tuple t, array n) / array n can be 'R', 'S', 'R' 'S', or 'S', 'R'34. 7     if n.size() = 1 and n1 = 'R'

47、;35. 8       Emit(tuple t, null)復(fù)制代碼分組聚合（GroupBy and Aggregation）分組聚合可以在如下的一個MapReduce中完成。Mapper抽取數(shù)據(jù)并將之分組聚合，Reducer 中對收到的數(shù)據(jù)再次聚合。典型的聚合應(yīng)用比如求和與最值可以以流的方式進行計算，因而不需要同時保有所有的值。但是另外一些情景就必須要兩階段MapReduce，前面提到過的惟一值模式就是一個這種類型的例子。size=14.166666030883789px36. 1 class Mapper37. 2   met

48、hod Map(null, tuple value GroupBy, value AggregateBy, value .)38. 3     Emit(value GroupBy, value AggregateBy)39. 4     40. 5 class Reducer41. 6   method Reduce(value GroupBy, v1, v2,.)42. 7     Emit(value GroupBy, aggregate( v1,

49、v2,. ) ) 43. 8     / aggregate() : sum(), max(),.復(fù)制代碼連接（Joining）MapperReduce框架可以很好地處理連接，不過在面對不同的數(shù)據(jù)量和處理效率要求的時候還是有一些技巧。在這部分我們會介紹一些基本方法，在后面的參考文檔中還列出了一些關(guān)于這方面的專題文章。分配后連接 （Reduce端連接,排序-合并連接）這個算法按照鍵K來連接數(shù)據(jù)集R和L。Mapper 遍歷R和L中的所有元組，以K為鍵輸出每一個標(biāo)記了來自于R還是L的元組，Reducer把同一個K的數(shù)據(jù)分裝入兩個容器（R和L），然后嵌套循環(huán)遍歷兩個容器中的數(shù)據(jù)以得到交集，最后輸出的每一條結(jié)果都包含了R中的數(shù)據(jù)、L中的數(shù)據(jù)和K。這種方法有以下缺點：· Mapper要輸出所有的數(shù)據(jù)，即使一些key只會在一個集合中出