電子發(fā)燒友App
前言
? ? ? ?本文總結(jié)了幾種網(wǎng)上或者論文中常見(jiàn)的MapReduce模式和算法，并系統(tǒng)化的解釋了這些技術(shù)的不同之處。所有描述性的文字和代碼都使用了標(biāo)準(zhǔn)hadoop的MapReduce模型，包括Mappers， Reduces， Combiners， Partitioners，和 sorting。詳細(xì)分析如下所示。
基本MapReduce模式
計(jì)數(shù)與求和
問(wèn)題陳述： 有許多文檔，每個(gè)文檔都有一些字段組成。需要計(jì)算出每個(gè)字段在所有文檔中的出現(xiàn)次數(shù)或者這些字段的其他什么統(tǒng)計(jì)值。例如，給定一個(gè)log文件，其中的每條記錄都包含一個(gè)響應(yīng)時(shí)間，需要計(jì)算出平均響應(yīng)時(shí)間。
解決方案：
讓我們先從簡(jiǎn)單的例子入手。在下面的代碼片段里，Mapper每遇到指定詞就把頻次記1，Reducer一個(gè)個(gè)遍歷這些詞的集合然后把他們的頻次加和。
?
這種方法的缺點(diǎn)顯而易見(jiàn)，Mapper提交了太多無(wú)意義的計(jì)數(shù)。它完全可以通過(guò)先對(duì)每個(gè)文檔中的詞進(jìn)行計(jì)數(shù)從而減少傳遞給Reducer的數(shù)據(jù)量：
?
如果要累計(jì)計(jì)數(shù)的的不只是單個(gè)文檔中的內(nèi)容，還包括了一個(gè)Mapper節(jié)點(diǎn)處理的所有文檔，那就要用到Combiner了：
?
應(yīng)用：
Log 分析， 數(shù)據(jù)查詢
整理歸類
問(wèn)題陳述：
有一系列條目，每個(gè)條目都有幾個(gè)屬性，要把具有同一屬性值的條目都保存在一個(gè)文件里，或者把條目按照屬性值分組。 最典型的應(yīng)用是倒排索引。
解決方案：
解決方案很簡(jiǎn)單。 在 Mapper 中以每個(gè)條目的所需屬性值作為 key，其本身作為值傳遞給 Reducer。 Reducer 取得按照屬性值分組的條目，然后可以處理或者保存。如果是在構(gòu)建倒排索引，那么 每個(gè)條目相當(dāng)于一個(gè)詞而屬性值就是詞所在的文檔ID。
應(yīng)用：
倒排索引， ETL
過(guò)濾 （文本查找），解析和校驗(yàn)
問(wèn)題陳述：
假設(shè)有很多條記錄，需要從其中找出滿足某個(gè)條件的所有記錄，或者將每條記錄傳換成另外一種形式（轉(zhuǎn)換操作相對(duì)于各條記錄獨(dú)立，即對(duì)一條記錄的操作與其他記錄無(wú)關(guān)）。像文本解析、特定值抽取、格式轉(zhuǎn)換等都屬于后一種用例。
解決方案：
非常簡(jiǎn)單，在Mapper 里逐條進(jìn)行操作，輸出需要的值或轉(zhuǎn)換后的形式。
應(yīng)用：
日志分析，數(shù)據(jù)查詢，ETL，數(shù)據(jù)校驗(yàn)
分布式任務(wù)執(zhí)行
問(wèn)題陳述：
大型計(jì)算可以分解為多個(gè)部分分別進(jìn)行然后合并各個(gè)計(jì)算的結(jié)果以獲得最終結(jié)果。
解決方案： 將數(shù)據(jù)切分成多份作為每個(gè) Mapper 的輸入，每個(gè)Mapper處理一份數(shù)據(jù)，執(zhí)行同樣的運(yùn)算，產(chǎn)生結(jié)果，Reducer把多個(gè)Mapper的結(jié)果組合成一個(gè)。
案例研究： 數(shù)字通信系統(tǒng)模擬
像 WiMAX 這樣的數(shù)字通信模擬軟件通過(guò)系統(tǒng)模型來(lái)傳輸大量的隨機(jī)數(shù)據(jù)，然后計(jì)算傳輸中的錯(cuò)誤幾率。 每個(gè) Mapper 處理樣本 1/N 的數(shù)據(jù)，計(jì)算出這部分?jǐn)?shù)據(jù)的錯(cuò)誤率，然后在 Reducer 里計(jì)算平均錯(cuò)誤率。
應(yīng)用：
工程模擬，數(shù)字分析，性能測(cè)試
排序
問(wèn)題陳述：
有許多條記錄，需要按照某種規(guī)則將所有記錄排序或是按照順序來(lái)處理記錄。
解決方案： 簡(jiǎn)單排序很好辦 – Mappers 將待排序的屬性值為鍵，整條記錄為值輸出。 不過(guò)實(shí)際應(yīng)用中的排序要更加巧妙一點(diǎn)， 這就是它之所以被稱為MapReduce 核心的原因（“核心”是說(shuō)排序？因?yàn)樽C明Hadoop計(jì)算能力的實(shí)驗(yàn)是大數(shù)據(jù)排序？還是說(shuō)Hadoop的處理過(guò)程中對(duì)key排序的環(huán)節(jié)？）。在實(shí)踐中，常用組合鍵來(lái)實(shí)現(xiàn)二次排序和分組。
MapReduce 最初只能夠?qū)︽I排序， 但是也有技術(shù)利用可以利用Hadoop 的特性來(lái)實(shí)現(xiàn)按值排序。想了解的話可以看 這篇博客。
按照BigTable的概念，使用 MapReduce來(lái)對(duì)最初數(shù)據(jù)而非中間數(shù)據(jù)排序，也即保持?jǐn)?shù)據(jù)的有序狀態(tài)更有好處，必須注意這一點(diǎn)。換句話說(shuō)，在數(shù)據(jù)插入時(shí)排序一次要比在每次查詢數(shù)數(shù)據(jù)的時(shí)候排序更高效。
應(yīng)用：
ETL，數(shù)據(jù)分析
#e#
非基本 MapReduce 模式
迭代消息傳遞 （圖處理）
問(wèn)題陳述：
假設(shè)一個(gè)實(shí)體網(wǎng)絡(luò)，實(shí)體之間存在著關(guān)系。 需要按照與它比鄰的其他實(shí)體的屬性計(jì)算出一個(gè)狀態(tài)。這個(gè)狀態(tài)可以表現(xiàn)為它和其它節(jié)點(diǎn)之間的距離， 存在特定屬性的鄰接點(diǎn)的跡象， 鄰域密度特征等等。
解決方案：
網(wǎng)絡(luò)存儲(chǔ)為系列節(jié)點(diǎn)的結(jié)合，每個(gè)節(jié)點(diǎn)包含有其所有鄰接點(diǎn)ID的列表。按照這個(gè)概念，MapReduce 迭代進(jìn)行，每次迭代中每個(gè)節(jié)點(diǎn)都發(fā)消息給它的鄰接點(diǎn)。鄰接點(diǎn)根據(jù)接收到的信息更新自己的狀態(tài)。當(dāng)滿足了某些條件的時(shí)候迭代停止，如達(dá)到了最大迭代次數(shù)（網(wǎng)絡(luò)半徑）或兩次連續(xù)的迭代幾乎沒(méi)有狀態(tài)改變。從技術(shù)上來(lái)看，Mapper 以每個(gè)鄰接點(diǎn)的ID為鍵發(fā)出信息，所有的信息都會(huì)按照接受節(jié)點(diǎn)分組，reducer 就能夠重算各節(jié)點(diǎn)的狀態(tài)然后更新那些狀態(tài)改變了的節(jié)點(diǎn)。下面展示了這個(gè)算法：
class Mapper
method Map（id n， object N）
Emit（id n， object N）
for all id m in N.OutgoingRelations do
Emit（id m， message getMessage（N））
class Reducer
method Reduce（id m， ［s1， s2，。..］）
M = null
messages = ［］
for all s in ［s1， s2，。..］ do
if IsObject（s） then
M = s
else // s is a message
messages.add（s）
M.State = calculateState（messages）
Emit（id m， item M）
一個(gè)節(jié)點(diǎn)的狀態(tài)可以迅速的沿著網(wǎng)絡(luò)傳全網(wǎng)，那些被感染了的節(jié)點(diǎn)又去感染它們的鄰居，整個(gè)過(guò)程就像下面的圖示一樣：
?
案例研究： 沿分類樹(shù)的有效性傳遞
問(wèn)題陳述：
這個(gè)問(wèn)題來(lái)自于真實(shí)的電子商務(wù)應(yīng)用。將各種貨物分類，這些類別可以組成一個(gè)樹(shù)形結(jié)構(gòu)，比較大的分類（像男人、女人、兒童）可以再分出小分類（像男褲或女裝），直到不能再分為止（像男式藍(lán)色牛仔褲）。這些不能再分的基層類別可以是有效（這個(gè)類別包含有貨品）或者已無(wú)效的（沒(méi)有屬于這個(gè)分類的貨品）。如果一個(gè)分類至少含有一個(gè)有效的子分類那么認(rèn)為這個(gè)分類也是有效的。我們需要在已知一些基層分類有效的情況下找出分類樹(shù)上所有有效的分類。
解決方案：
這個(gè)問(wèn)題可以用上一節(jié)提到的框架來(lái)解決。我們咋下面定義了名為 getMessage和 calculateState 的方法：
class N
State in {True = 2， False = 1， null = 0}，
initialized 1 or 2 for end-of-line categories， 0 otherwise
method getMessage（object N）
return N.State
method calculateState（state s， data ［d1， d2，。..］）
return max（ ［d1， d2，。..］ ）
案例研究：廣度優(yōu)先搜索
問(wèn)題陳述：需要計(jì)算出一個(gè)圖結(jié)構(gòu)中某一個(gè)節(jié)點(diǎn)到其它所有節(jié)點(diǎn)的距離。
解決方案： Source源節(jié)點(diǎn)給所有鄰接點(diǎn)發(fā)出值為0的信號(hào)，鄰接點(diǎn)把收到的信號(hào)再轉(zhuǎn)發(fā)給自己的鄰接點(diǎn)，每轉(zhuǎn)發(fā)一次就對(duì)信號(hào)值加1：
class N
State is distance，
initialized 0 for source node， INFINITY for all other nodes
method getMessage（N）
return N.State + 1
method calculateState（state s， data ［d1， d2，。..］）
min（ ［d1， d2，。..］ ）
案例研究：網(wǎng)頁(yè)排名和 Mapper 端數(shù)據(jù)聚合
這個(gè)算法由Google提出，使用權(quán)威的PageRank算法，通過(guò)連接到一個(gè)網(wǎng)頁(yè)的其他網(wǎng)頁(yè)來(lái)計(jì)算網(wǎng)頁(yè)的相關(guān)性。真實(shí)算法是相當(dāng)復(fù)雜的，但是核心思想是權(quán)重可以傳播，也即通過(guò)一個(gè)節(jié)點(diǎn)的各聯(lián)接節(jié)點(diǎn)的權(quán)重的均值來(lái)計(jì)算節(jié)點(diǎn)自身的權(quán)重。
class N
State is PageRank
method getMessage（object N）
return N.State / N.OutgoingRelations.size（）
method calculateState（state s， data ［d1， d2，。..］）
return （ sum（［d1， d2，。..］） ）
要指出的是上面用一個(gè)數(shù)值來(lái)作為評(píng)分實(shí)際上是一種簡(jiǎn)化，在實(shí)際情況下，我們需要在Mapper端來(lái)進(jìn)行聚合計(jì)算得出這個(gè)值。下面的代碼片段展示了這個(gè)改變后的邏輯 （針對(duì)于 PageRank 算法）：
class Mapper
method Initialize
H = new AssociativeArray
method Map（id n， object N）
p = N.PageRank / N.OutgoingRelations.size（）
Emit（id n， object N）
for all id m in N.OutgoingRelations do
H{m} = H{m} + p
method Close
for all id n in H do
Emit（id n， value H{n}）
class Reducer
method Reduce（id m， ［s1， s2，。..］）
M = null
p = 0
for all s in ［s1， s2，。..］ do
if IsObject（s） then
M = s
else
p = p + s
M.PageRank = p
Emit（id m， item M）
應(yīng)用：
圖分析，網(wǎng)頁(yè)索引
#e#
值去重 （對(duì)唯一項(xiàng)計(jì)數(shù)）
問(wèn)題陳述： 記錄包含值域F和值域 G，要分別統(tǒng)計(jì)相同G值的記錄中不同的F值的數(shù)目 （相當(dāng)于按照 G分組）。
這個(gè)問(wèn)題可以推而廣之應(yīng)用于分面搜索（某些電子商務(wù)網(wǎng)站稱之為Narrow Search）
Record 1： F=1， G={a， b}
Record 2： F=2， G={a， d， e}
Record 3： F=1， G=
Record 4： F=3， G={a， b}
Result：
a -》 3 // F=1， F=2， F=3
b -》 2 // F=1， F=3
d -》 1 // F=2
e -》 1 // F=2
解決方案 I：
第一種方法是分兩個(gè)階段來(lái)解決這個(gè)問(wèn)題。第一階段在Mapper中使用F和G組成一個(gè)復(fù)合值對(duì)，然后在Reducer中輸出每個(gè)值對(duì)，目的是為了保證F值的唯一性。在第二階段，再將值對(duì)按照G值來(lái)分組計(jì)算每組中的條目數(shù)。
第一階段：
class Mapper
method Map（null， record ［value f， categories ［g1， g2，。..］］）
for all category g in ［g1， g2，。..］
Emit（record ［g， f］， count 1）
class Reducer
method Reduce（record ［g， f］， counts ［n1， n2， 。..］）
Emit（record ［g， f］， null ）
第二階段：
class Mapper
method Map（record ［f， g］， null）
Emit（value g， count 1）
class Reducer
method Reduce（value g， counts ［n1， n2，。..］）
Emit（value g， sum（ ［n1， n2，。..］ ） ）
解決方案 II：
第二種方法只需要一次MapReduce 即可實(shí)現(xiàn)，但擴(kuò)展性不強(qiáng)。算法很簡(jiǎn)單-Mapper 輸出值和分類，在Reducer里為每個(gè)值對(duì)應(yīng)的分類去重然后給每個(gè)所屬的分類計(jì)數(shù)加1，最后再在Reducer結(jié)束后將所有計(jì)數(shù)加和。這種方法適用于只有有限個(gè)分類，而且擁有相同F(xiàn)值的記錄不是很多的情況。例如網(wǎng)絡(luò)日志處理和用戶分類，用戶的總數(shù)很多，但是每個(gè)用戶的事件是有限的，以此分類得到的類別也是有限的。值得一提的是在這種模式下可以在數(shù)據(jù)傳輸?shù)絉educer之前使用Combiner來(lái)去除分類的重復(fù)值。
class Mapper
method Map（null， record ［value f， categories ［g1， g2，。..］ ）
for all category g in ［g1， g2，。..］
Emit（value f， category g）
class Reducer
method Initialize
H = new AssociativeArray ： category -》 count
method Reduce（value f， categories ［g1， g2，。..］）
［g1‘， g2’，。.］ = ExcludeDuplicates（ ［g1， g2，。.］ ）
for all category g in ［g1‘， g2’，。..］
H{g} = H{g} + 1
method Close
for all category g in H do
Emit（category g， count H{g}）
應(yīng)用：
日志分析，用戶計(jì)數(shù)
互相關(guān)
問(wèn)題陳述：有多個(gè)各由若干項(xiàng)構(gòu)成的組，計(jì)算項(xiàng)兩兩共同出現(xiàn)于一個(gè)組中的次數(shù)。假如項(xiàng)數(shù)是N，那么應(yīng)該計(jì)算N*N。
這種情況常見(jiàn)于文本分析（條目是單詞而元組是句子），市場(chǎng)分析（購(gòu)買了此物的客戶還可能購(gòu)買什么）。如果N*N小到可以容納于一臺(tái)機(jī)器的內(nèi)存，實(shí)現(xiàn)起來(lái)就比較簡(jiǎn)單了。
配對(duì)法
第一種方法是在Mapper中給所有條目配對(duì)，然后在Reducer中將同一條目對(duì)的計(jì)數(shù)加和。但這種做法也有缺點(diǎn)：
· 使用 combiners 帶來(lái)的的好處有限，因?yàn)楹芸赡芩许?xiàng)對(duì)都是唯一的
· 不能有效利用內(nèi)存
class Mapper
method Map（null， items ［i1， i2，。..］ ）
for all item i in ［i1， i2，。..］
for all item j in ［i1， i2，。..］
Emit（pair ［i j］， count 1）
class Reducer
method Reduce（pair ［i j］， counts ［c1， c2，。..］）
s = sum（［c1， c2，。..］）
Emit（pair［i j］， count s）
Stripes Approach（條方法？不知道這個(gè)名字怎么理解）
第二種方法是將數(shù)據(jù)按照pair中的第一項(xiàng)來(lái)分組，并維護(hù)一個(gè)關(guān)聯(lián)數(shù)組，數(shù)組中存儲(chǔ)的是所有關(guān)聯(lián)項(xiàng)的計(jì)數(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é)果的鍵數(shù)量相對(duì)較少，因此減少了排序消耗。
· 可以有效利用 combiners。
· 可在內(nèi)存中執(zhí)行，不過(guò)如果沒(méi)有正確執(zhí)行的話也會(huì)帶來(lái)問(wèn)題。
· 實(shí)現(xiàn)起來(lái)比較復(fù)雜。
· 一般來(lái)說(shuō)， “stripes” 比 “pairs” 更快
class Mapper
method Map（null， items ［i1， i2，。..］ ）
for all item i in ［i1， i2，。..］
H = new AssociativeArray ： item -》 counter
for all item j in ［i1， i2，。..］
H{j} = H{j} + 1
Emit（item i， stripe H）
class Reducer
method Reduce（item i， stripes ［H1， H2，。..］）
H = new AssociativeArray ： item -》 counter
H = merge-sum（ ［H1， H2，。..］ ）
for all item j in H.keys（）
Emit（pair ［i j］， H{j}）
應(yīng)用：
文本分析，市場(chǎng)分析
References：
1. Lin J. Dyer C. Hirst G. Data Intensive Processing MapReduce
用MapReduce 表達(dá)關(guān)系模式
在這部分我們會(huì)討論一下怎么使用MapReduce來(lái)進(jìn)行主要的關(guān)系操作。
篩選（Selection）
class Mapper
method Map（rowkey key， tuple t）
if t satisfies the predicate
Emit（tuple t， null）
投影（Projection）
投影只比篩選稍微復(fù)雜一點(diǎn)，在這種情況下我們可以用Reducer來(lái)消除可能的重復(fù)值
class Mapper
method Map（rowkey key， tuple t）
tuple g = project（t） // extract required fields to tuple g
Emit（tuple g， null）
class Reducer
method Reduce（tuple t， array n） // n is an array of nulls
Emit（tuple t， null）
合并（Union）
兩個(gè)數(shù)據(jù)集中的所有記錄都送入Mapper，在Reducer里消重。
class Mapper
method Map（rowkey key， tuple t）
Emit（tuple t， null）
class Reducer
method Reduce（tuple t， array n） // n is an array of one or two nulls
Emit（tuple t， null）
交集（Intersection）
將兩個(gè)數(shù)據(jù)集中需要做交叉的記錄輸入Mapper，Reducer 輸出出現(xiàn)了兩次的記錄。因?yàn)槊織l記錄都有一個(gè)主鍵，在每個(gè)數(shù)據(jù)集中只會(huì)出現(xiàn)一次，所以這樣做是可行的。
差異（Difference）
假設(shè)有兩個(gè)數(shù)據(jù)集R和S，我們要找出R與S的差異。Mapper將所有的元組做上標(biāo)記，表明他們來(lái)自于R還是S，Reducer只輸出那些存在于R中而不在S中的記錄。
class Mapper
method Map（rowkey key， tuple t）
Emit（tuple t， string t.SetName） // t.SetName is either ‘R’ or ‘S’
class Reducer
method Reduce（tuple t， array n） // array n can be ［‘R’］， ［‘S’］， ［‘R’ ‘S’］， or ［‘S’， ‘R’］
if n.size（） = 1 and n［1］ = ‘R’
Emit（tuple t， null）
分組聚合（GroupBy and Aggregation）
分組聚合可以在如下的一個(gè)MapReduce中完成。Mapper抽取數(shù)據(jù)并將之分組聚合，Reducer 中對(duì)收到的數(shù)據(jù)再次聚合。典型的聚合應(yīng)用比如求和與最值可以以流的方式進(jìn)行計(jì)算，因而不需要同時(shí)保有所有的值。但是另外一些情景就必須要兩階段MapReduce，前面提到過(guò)的惟一值模式就是一個(gè)這種類型的例子。
連接（Joining）
MapperReduce框架可以很好地處理連接，不過(guò)在面對(duì)不同的數(shù)據(jù)量和處理效率要求的時(shí)候還是有一些技巧。在這部分我們會(huì)介紹一些基本方法，在后面的參考文檔中還列出了一些關(guān)于這方面的專題文章。
分配后連接 （Reduce端連接，排序-合并連接）
這個(gè)算法按照鍵K來(lái)連接數(shù)據(jù)集R和L。Mapper 遍歷R和L中的所有元組，以K為鍵輸出每一個(gè)標(biāo)記了來(lái)自于R還是L的元組，Reducer把同一個(gè)K的數(shù)據(jù)分裝入兩個(gè)容器（R和L），然后嵌套循環(huán)遍歷兩個(gè)容器中的數(shù)據(jù)以得到交集，最后輸出的每一條結(jié)果都包含了R中的數(shù)據(jù)、L中的數(shù)據(jù)和K。這種方法有以下缺點(diǎn)：
· Mapper要輸出所有的數(shù)據(jù)，即使一些key只會(huì)在一個(gè)集合中出現(xiàn)。
· Reducer 要在內(nèi)存中保有一個(gè)key的所有數(shù)據(jù)，如果數(shù)據(jù)量打過(guò)了內(nèi)存，那么就要緩存到硬盤上，這就增加了硬盤IO的消耗。
盡管如此，再分配連接方式仍然是最通用的方法，特別是其他優(yōu)化技術(shù)都不適用的時(shí)候。
class Mapper
method Map（null， tuple ［join_key k， value v1， value v2，。..］）
Emit（join_key k， tagged_tuple ［set_name tag， values ［v1， v2， 。..］ ］ ）
class Reducer
method Reduce（join_key k， tagged_tuples ［t1， t2，。..］）
H = new AssociativeArray ： set_name -》 values
for all tagged_tuple t in ［t1， t2，。..］ // separate values into 2 arrays
H{t.tag}.add（t.values）
for all values r in H{‘R’} // produce a cross-join of the two arrays
for all values l in H{‘L’}
Emit（null， ［k r l］ ）
復(fù)制鏈接Replicated Join （Mapper端連接， Hash 連接）
在實(shí)際應(yīng)用中，將一個(gè)小數(shù)據(jù)集和一個(gè)大數(shù)據(jù)集連接是很常見(jiàn)的（如用戶與日志記錄）。假定要連接兩個(gè)集合R和L，其中R相對(duì)較小，這樣，可以把R分發(fā)給所有的Mapper，每個(gè)Mapper都可以載入它并以連接鍵來(lái)索引其中的數(shù)據(jù)，最常用和有效的索引技術(shù)就是哈希表。之后，Mapper遍歷L，并將其與存儲(chǔ)在哈希表中的R中的相應(yīng)記錄連接，。這種方法非常高效，因?yàn)椴恍枰獙?duì)L中的數(shù)據(jù)排序，也不需要通過(guò)網(wǎng)絡(luò)傳送L中的數(shù)據(jù)，但是R必須足夠小到能夠分發(fā)給所有的Mapper。
class Mapper
method Initialize
H = new AssociativeArray ： join_key -》 tuple from R
R = loadR（）
for all ［ join_key k， tuple ［r1， r2，。..］ ］ in R
H{k} = H{k}.append（ ［r1， r2，。..］ ）
method Map（join_key k， tuple l）
for all tuple r in H{k}
Emit（null， tuple ［k r l］ ）
工商網(wǎng)監(jiān)
評(píng)論