谷歌技術(shù)"三寶"之MapReduce

oskycar 2014-12-04

展開全文

江湖傳說永流傳：谷歌技術(shù)有"三寶"，GFS、MapReduce和大表（BigTable）！

谷歌在03到06年間連續(xù)發(fā)表了三篇很有影響力的文章，分別是03年SOSP的GFS，04年OSDI的MapReduce，和06年OSDI的BigTable。SOSP和OSDI都是操作系統(tǒng)領(lǐng)域的頂級會議，在計算機學(xué)會推薦會議里屬于A類。SOSP在單數(shù)年舉辦，而OSDI在雙數(shù)年舉辦。

那么這篇博客就來介紹一下MapReduce。

1. MapReduce是干啥的

因為沒找到谷歌的示意圖，所以我想借用一張Hadoop項目的結(jié)構(gòu)圖來說明下MapReduce所處的位置，如下圖。

Hadoop實際上就是谷歌三寶的開源實現(xiàn)，Hadoop MapReduce對應(yīng)Google MapReduce，HBase對應(yīng)BigTable，HDFS對應(yīng)GFS。HDFS（或GFS）為上層提供高效的非結(jié)構(gòu)化存儲服務(wù)，HBase（或BigTable）是提供結(jié)構(gòu)化數(shù)據(jù)服務(wù)的分布式數(shù)據(jù)庫，Hadoop MapReduce（或Google MapReduce）是一種并行計算的編程模型，用于作業(yè)調(diào)度。

GFS和BigTable已經(jīng)為我們提供了高性能、高并發(fā)的服務(wù)，但是并行編程可不是所有程序員都玩得轉(zhuǎn)的活兒，如果我們的應(yīng)用本身不能并發(fā)，那GFS、BigTable也都是沒有意義的。MapReduce的偉大之處就在于讓不熟悉并行編程的程序員也能充分發(fā)揮分布式系統(tǒng)的威力。

簡單概括的說，MapReduce是將一個大作業(yè)拆分為多個小作業(yè)的框架（大作業(yè)和小作業(yè)應(yīng)該本質(zhì)是一樣的，只是規(guī)模不同），用戶需要做的就是決定拆成多少份，以及定義作業(yè)本身。

下面用一個貫穿全文的例子來解釋MapReduce是如何工作的。

2. 例子：統(tǒng)計詞頻

如果我想統(tǒng)計下過去10年計算機論文出現(xiàn)最多的幾個單詞，看看大家都在研究些什么，那我收集好論文后，該怎么辦呢？

方法一：我可以寫一個小程序，把所有論文按順序遍歷一遍，統(tǒng)計每一個遇到的單詞的出現(xiàn)次數(shù)，最后就可以知道哪幾個單詞最熱門了。

這種方法在數(shù)據(jù)集比較小時，是非常有效的，而且實現(xiàn)最簡單，用來解決這個問題很合適。

方法二：寫一個多線程程序，并發(fā)遍歷論文。

這個問題理論上是可以高度并發(fā)的，因為統(tǒng)計一個文件時不會影響統(tǒng)計另一個文件。當(dāng)我們的機器是多核或者多處理器，方法二肯定比方法一高效。但是寫一個多線程程序要比方法一困難多了，我們必須自己同步共享數(shù)據(jù)，比如要防止兩個線程重復(fù)統(tǒng)計文件。

方法三：把作業(yè)交給多個計算機去完成。

我們可以使用方法一的程序，部署到N臺機器上去，然后把論文集分成N份，一臺機器跑一個作業(yè)。這個方法跑得足夠快，但是部署起來很麻煩，我們要人工把程序copy到別的機器，要人工把論文集分開，最痛苦的是還要把N個運行結(jié)果進行整合（當(dāng)然我們也可以再寫一個程序）。

方法四：讓MapReduce來幫幫我們吧！

MapReduce本質(zhì)上就是方法三，但是如何拆分文件集，如何copy程序，如何整合結(jié)果這些都是框架定義好的。我們只要定義好這個任務(wù)（用戶程序），其它都交給MapReduce。

在介紹MapReduce如何工作之前，先講講兩個核心函數(shù)map和reduce以及MapReduce的偽代碼。

3. map函數(shù)和reduce函數(shù)

map函數(shù)和reduce函數(shù)是交給用戶實現(xiàn)的，這兩個函數(shù)定義了任務(wù)本身。

map函數(shù)：接受一個鍵值對（key-value pair），產(chǎn)生一組中間鍵值對。MapReduce框架會將map函數(shù)產(chǎn)生的中間鍵值對里鍵相同的值傳遞給一個reduce函數(shù)。
reduce函數(shù)：接受一個鍵，以及相關(guān)的一組值，將這組值進行合并產(chǎn)生一組規(guī)模更小的值（通常只有一個或零個值）。

統(tǒng)計詞頻的MapReduce函數(shù)的核心代碼非常簡短，主要就是實現(xiàn)這兩個函數(shù)。

[plain] view plain copy print ?

map(String key, String value):
// key: document name
// value: document contents
for each word w in value:
EmitIntermediate(w, "1");
reduce(String key, Iterator values):
// key: a word
// values: a list of counts
int result = 0;
for each v in values:
result += ParseInt(v);
Emit(AsString(result));

在統(tǒng)計詞頻的例子里，map函數(shù)接受的鍵是文件名，值是文件的內(nèi)容，map逐個遍歷單詞，每遇到一個單詞w，就產(chǎn)生一個中間鍵值對<w, "1">，這表示單詞w咱又找到了一個；MapReduce將鍵相同（都是單詞w）的鍵值對傳給reduce函數(shù)，這樣reduce函數(shù)接受的鍵就是單詞w，值是一串"1"（最基本的實現(xiàn)是這樣，但可以優(yōu)化），個數(shù)等于鍵為w的鍵值對的個數(shù)，然后將這些“1”累加就得到單詞w的出現(xiàn)次數(shù)。最后這些單詞的出現(xiàn)次數(shù)會被寫到用戶定義的位置，存儲在底層的分布式存儲系統(tǒng)（GFS或HDFS）。

4. MapReduce是如何工作的

上圖是論文里給出的流程圖。一切都是從最上方的user program開始的，user program鏈接了MapReduce庫，實現(xiàn)了最基本的Map函數(shù)和Reduce函數(shù)。圖中執(zhí)行的順序都用數(shù)字標(biāo)記了。

MapReduce庫先把user program的輸入文件劃分為M份（M為用戶定義），每一份通常有16MB到64MB，如圖左方所示分成了split0~4；然后使用fork將用戶進程拷貝到集群內(nèi)其它機器上。
user program的副本中有一個稱為master，其余稱為worker，master是負(fù)責(zé)調(diào)度的，為空閑worker分配作業(yè)（Map作業(yè)或者Reduce作業(yè)），worker的數(shù)量也是可以由用戶指定的。
被分配了Map作業(yè)的worker，開始讀取對應(yīng)分片的輸入數(shù)據(jù)，Map作業(yè)數(shù)量是由M決定的，和split一一對應(yīng)；Map作業(yè)從輸入數(shù)據(jù)中抽取出鍵值對，每一個鍵值對都作為參數(shù)傳遞給map函數(shù)，map函數(shù)產(chǎn)生的中間鍵值對被緩存在內(nèi)存中。
緩存的中間鍵值對會被定期寫入本地磁盤，而且被分為R個區(qū)，R的大小是由用戶定義的，將來每個區(qū)會對應(yīng)一個Reduce作業(yè)；這些中間鍵值對的位置會被通報給master，master負(fù)責(zé)將信息轉(zhuǎn)發(fā)給Reduce worker。
master通知分配了Reduce作業(yè)的worker它負(fù)責(zé)的分區(qū)在什么位置（肯定不止一個地方，每個Map作業(yè)產(chǎn)生的中間鍵值對都可能映射到所有R個不同分區(qū)），當(dāng)Reduce worker把所有它負(fù)責(zé)的中間鍵值對都讀過來后，先對它們進行排序，使得相同鍵的鍵值對聚集在一起。因為不同的鍵可能會映射到同一個分區(qū)也就是同一個Reduce作業(yè)（誰讓分區(qū)少呢），所以排序是必須的。
reduce worker遍歷排序后的中間鍵值對，對于每個唯一的鍵，都將鍵與關(guān)聯(lián)的值傳遞給reduce函數(shù)，reduce函數(shù)產(chǎn)生的輸出會添加到這個分區(qū)的輸出文件中。
當(dāng)所有的Map和Reduce作業(yè)都完成了，master喚醒正版的user program，MapReduce函數(shù)調(diào)用返回user program的代碼。

所有執(zhí)行完畢后，MapReduce輸出放在了R個分區(qū)的輸出文件中（分別對應(yīng)一個Reduce作業(yè)）。用戶通常并不需要合并這R個文件，而是將其作為輸入交給另一個MapReduce程序處理。整個過程中，輸入數(shù)據(jù)是來自底層分布式文件系統(tǒng)（GFS）的，中間數(shù)據(jù)是放在本地文件系統(tǒng)的，最終輸出數(shù)據(jù)是寫入底層分布式文件系統(tǒng)（GFS）的。而且我們要注意Map/Reduce作業(yè)和map/reduce函數(shù)的區(qū)別：Map作業(yè)處理一個輸入數(shù)據(jù)的分片，可能需要調(diào)用多次map函數(shù)來處理每個輸入鍵值對；Reduce作業(yè)處理一個分區(qū)的中間鍵值對，期間要對每個不同的鍵調(diào)用一次reduce函數(shù)，Reduce作業(yè)最終也對應(yīng)一個輸出文件。

我更喜歡把流程分為三個階段。第一階段是準(zhǔn)備階段，包括1、2，主角是MapReduce庫，完成拆分作業(yè)和拷貝用戶程序等任務(wù)；第二階段是運行階段，包括3、4、5、6，主角是用戶定義的map和reduce函數(shù)，每個小作業(yè)都獨立運行著；第三階段是掃尾階段，這時作業(yè)已經(jīng)完成，作業(yè)結(jié)果被放在輸出文件里，就看用戶想怎么處理這些輸出了。

5. 詞頻是怎么統(tǒng)計出來的

結(jié)合第四節(jié)，我們就可以知道第三節(jié)的代碼是如何工作的了。假設(shè)咱們定義M=5，R=3，并且有6臺機器，一臺master。

這幅圖描述了MapReduce如何處理詞頻統(tǒng)計。由于map worker數(shù)量不夠，首先處理了分片1、3、4，并產(chǎn)生中間鍵值對；當(dāng)所有中間值都準(zhǔn)備好了，Reduce作業(yè)就開始讀取對應(yīng)分區(qū)，并輸出統(tǒng)計結(jié)果。

6. 用戶的權(quán)利

用戶最主要的任務(wù)是實現(xiàn)map和reduce接口，但還有一些有用的接口是向用戶開放的。

an input reader。這個函數(shù)會將輸入分為M個部分，并且定義了如何從數(shù)據(jù)中抽取最初的鍵值對，比如詞頻的例子中定義文件名和文件內(nèi)容是鍵值對。
a partition function。這個函數(shù)用于將map函數(shù)產(chǎn)生的中間鍵值對映射到一個分區(qū)里去，最簡單的實現(xiàn)就是將鍵求哈希再對R取模。
a compare function。這個函數(shù)用于Reduce作業(yè)排序，這個函數(shù)定義了鍵的大小關(guān)系。
an output writer。負(fù)責(zé)將結(jié)果寫入底層分布式文件系統(tǒng)。
a combiner function。實際就是reduce函數(shù)，這是用于前面提到的優(yōu)化的，比如統(tǒng)計詞頻時，如果每個<w, "1">要讀一次，因為reduce和map通常不在一臺機器，非常浪費時間，所以可以在map執(zhí)行的地方先運行一次combiner，這樣reduce只需要讀一次<w, "n">了。
map和reduce函數(shù)就不多說了。

7. MapReduce的實現(xiàn)

目前MapReduce已經(jīng)有多種實現(xiàn)，除了谷歌自己的實現(xiàn)外，還有著名的hadoop，區(qū)別是谷歌是c++，而hadoop是用java。另外斯坦福大學(xué)實現(xiàn)了一個在多核/多處理器、共享內(nèi)存環(huán)境內(nèi)運行的MapReduce，稱為Phoenix（介紹），相關(guān)的論文發(fā)表在07年的HPCA，是當(dāng)年的最佳論文哦！