1、MapReduce： 大規(guī)模集群上的簡化數(shù)據(jù)處理【摘要】 MapReduce是一個“與處理以及生成大量數(shù)據(jù)集相關聯(lián)的”程序模型。用戶通過定義一個map函數(shù)，處理鍵值對以生成一個中間鍵值對的集合，以及一個叫做reduce的函數(shù)用以合并所有先前map過后的有相同鍵的中間量?，F(xiàn)實世界中的許多任務在這個模型中得到了很好的表達，如下文所述。 程序員用這種風格的程序?qū)懗龅拇a可以自動并行以及在商用極其上大規(guī)模的處理數(shù)據(jù)。運行時系統(tǒng)關注輸入數(shù)據(jù)的分區(qū)，通過一系列機器的集合來規(guī)劃程序的執(zhí)行， 處理程序失效以及把控必要的系統(tǒng)內(nèi)部交互。這個框架的優(yōu)勢在于使得程序員無需任何并行與分布式系統(tǒng)的經(jīng)驗就可以容易的掌控大型

2、分布式系統(tǒng)的資源。 我們的MapReduce的實現(xiàn)是運行在商用機器的大規(guī)模集群之上，且擁有高可擴展性：一個典型的MapReduce運行場景是在數(shù)千臺機器上處理TB級數(shù)據(jù)。程序與系統(tǒng)易于使用：數(shù)百個MapReduce程序?qū)嵤┝藬?shù)千份的MapReduce的每天都運行于谷歌集群之上job。1.說明 在過去的五年里， 作者以及其它許多谷歌人開發(fā)出了大量的用于特定目的的處理大規(guī)模原始數(shù)據(jù)的程序，諸如網(wǎng)頁請求日志等等計算不同類型的派生數(shù)據(jù)，這些派生數(shù)據(jù)有網(wǎng)絡文件的迥異的圖形結(jié)構(gòu)，每個主機的頁面數(shù)，某一天所提出的最高頻率問題等等。大多數(shù)這種計算在概念上是想當容易把握的。然而，輸入的數(shù)據(jù)通常是大規(guī)模的且計算不

3、得不分布在數(shù)百甚至上千太的機器以在一個合理的時間內(nèi)得出計算結(jié)果。 有關如何并行計算，分布數(shù)據(jù)，處理失效則大大增加了原初直觀問題的難度。這就需要復雜的大量的代碼來處理這些問題。 作為對復雜性的反應，我們設計了一個全新的抽象模型允許表達簡潔的且為我們所需要的計算，同時該抽象模型屏蔽了大量底層并行細節(jié)，容錯處理， 數(shù)據(jù)分布，負載平衡（將這些設計細節(jié)放置于庫中）。我們的抽象模型的map 和 reduce最初靈感是來自于lisp這樣的諸多函數(shù)式編程語言。我們認識到絕大多數(shù)的計算關涉到“之于每一輸入邏輯記錄用于操作下一步的臨時鍵值對的map操作”，這些計算的下一步就是運用reduce操作操作所有共享同一個

4、鍵的變量，以恰當?shù)暮喜⒎植紨?shù)據(jù)。我們使用用戶定義的map和reduce函數(shù)模型，這一點使得我們得以對大規(guī)模計算并行化，同時重新執(zhí)行重要的容錯處理機制.該抽象模型的主要工作是形成了一個簡單而強有力的接口，該接口允許并行自動化以及使得分布在通用PC之上的大尺度集群計算成為可能. 第二部分描述了基礎的編程模型并給出了數(shù)個例子。第三部分描述了合適我們集群計算環(huán)境的MapReduce接口實現(xiàn)。第四部分給出了數(shù)個精巧的我們認為有用的程序模型。第五部分是各種不同任務下的性能測試。第六部分探索了運用MapReduce重寫谷歌產(chǎn)品檢索系統(tǒng)的情況。第七部分給出相關的討論以及未來的工作。2.程序模型 計算任務是這樣

5、描述的：有一個鍵值對的集合作為輸入，以及另外一個鍵值對的集合作為輸出。MapReduce 庫的使用者將計算抽象表達為兩種功能：Map和Reduce.Map由用戶書寫， 將鍵值對的一個集合作為輸入，同時產(chǎn)出一個作為中間狀態(tài)的鍵值對。MapReduce庫將所有擁有相同的鍵( 鍵I)的中間狀態(tài)鍵合并起來傳遞到Reduce功能。Reduce功能同樣是由用戶書寫的，它接收一個中間鍵( 鍵I)以及有關這個鍵的鍵值集合，將這些值進行歸并 merge以形成可能的更小的值集合。典型的場景是每一個Reduce操作所喚起的只是0個或者1個最終輸出值（鍵值）。中間值是由用戶的reduce功能使用的迭代器來支持的。這就

6、允許我們對相對于內(nèi)存來說過大的值隊列進行處理。2.1例子 考慮這樣一個問題：以一大堆文檔為單位查找某個詞語出現(xiàn)的詞頻。用戶可能象下面一樣書寫偽代碼：map(String key, String value):/ key: document name/ value: document contentsfor each word w in value:EmitIntermediate(w, 1);reduce(String key, Iterator values):/ key: a word/ values: a list of countsint result = 0;for each v i

7、n values:result += ParseInt(v);Emit(AsString(result);map功能就產(chǎn)出單詞，并對其出現(xiàn)的次數(shù)統(tǒng)計。偽代碼中的EmitIntermediate(w, 1)指的是出現(xiàn)1次，這里是特例。Reduce功能統(tǒng)計每一個特定單詞的出現(xiàn)次數(shù)。 用戶首先在mapreduce特定對象中結(jié)合輸入輸出文件編寫代碼，調(diào)整參數(shù)；其后喚醒MapReduce 功能， 將特定對象傳遞給它。用戶代碼是和MapReduce庫相綁定的（由C+ 語言實現(xiàn)）。附錄A包含了2.1例子的完整程序文本。2.2類型 即便前例偽代碼展示的是輸入輸出字符串，概念上由用戶定義的map和reduce功

8、能還支持以下類型：map (k1,v1) - list(k2,v2)reduce (k2,list(v2) - list(v2)也既，輸入的鍵值是從相對于輸出鍵值的不同域中提取，進一步的，中間鍵值同輸出鍵值都是來自于同一個域。我們用C+實現(xiàn)了用戶定義功能下的任意的字符串，并將置于用戶代碼的管轄，方便在string和任意類型之間轉(zhuǎn)換。2.3 更多的例子 這里展示了更多的簡單程序，這些小程序很清晰易懂的表達了MapReduce的計算。分布式grep: map 功能在匹配一給定模式時會產(chǎn)生一行。同時reduce 功能是一種身份揀選機制，將中間鍵值對拷貝操作后轉(zhuǎn)化為輸出鍵值對。URL接觸詞頻統(tǒng)計：ma

9、p功能處理網(wǎng)頁日志請求以及輸出 。reduce 功能將所有屬于同一個URL的值進行疊加并輸出 三元組。逆轉(zhuǎn)網(wǎng)絡鏈接圖：map功能為每一個在命名了資源的頁中能鏈接至target 的URL輸出鍵值對。reduce功能將所有關聯(lián)于一給定的target URL的源URL列表進行拼接并產(chǎn)出鍵值對。每主機的詞匯向量：一個詞匯/向量總結(jié)了出現(xiàn)在一個文檔中的最重要單詞，或者總結(jié)了出現(xiàn)在一個文檔集合中的最重要單詞，這個文檔集合被表示成 列表。map功能對每個輸入的文檔產(chǎn)生鍵值對（這里的主機名hostname是從輸入文檔的URL中提取的）。reduce 功能被傳遞了“對給定主機的所有的前文檔詞匯向量”。它將這些詞

10、匯變量加總起來，丟棄非高頻詞匯并最終產(chǎn)生鍵值對。反轉(zhuǎn)索引：map功能解析每一個文檔， 同時產(chǎn)出的一個序列。 reduce 功能接收對一給定單詞的所有鍵值對，對相應的文檔ID排序，同時產(chǎn)生鍵值對 所有輸出的鍵值對的集合形成了簡單的反轉(zhuǎn)索引。通過追蹤單詞的位值，這種計算是容易增加的（augment）分布式排序：map功能將每個鍵從記錄中提取出來，產(chǎn)生 鍵值對。reduce 功能產(chǎn)生所有未經(jīng)改變過的鍵值對。這種計算是依賴與4.1節(jié)所述之設備分區(qū)，以及4.2節(jié)所述次序?qū)傩浴?.實現(xiàn) MapReduce 接口的多種不同實現(xiàn)是可能的。依賴于環(huán)境作出正確的接口實現(xiàn)選擇。比如， 其中一種MapReduce 接

11、口可能的實現(xiàn)適用于小型共享內(nèi)存機器， 另一種MapReduce 接口實現(xiàn)則適用于多處理器NUMA, 還有可能的一種MapReduce 接口的適用情況是大規(guī)模網(wǎng)絡機器的集合。 這一節(jié)描述了以谷歌內(nèi)廣泛使用的計算環(huán)境為背景的一種MapReduce 接口實現(xiàn)：由轉(zhuǎn)換以太網(wǎng)鏈接起來的商用PC組成的大規(guī)模集群。典型的單臺機器硬件環(huán)境是x86雙核處理器，2到4GB內(nèi)存，操作系統(tǒng)為linux。商用網(wǎng)絡硬件典型在機器級別為100M/s或1G/s，但均值通常明顯低于帶寬的一半。一個集群由由數(shù)百或數(shù)千臺機器構(gòu)成，這種架構(gòu)必然帶來機器失效failure的情況。存儲設備由不怎么昂貴的IDE磁盤組成（同臺式電腦）。一個

12、谷歌內(nèi)部開發(fā)的分布式系統(tǒng)8用于管理存儲于這些機器磁盤上的數(shù)據(jù)。用戶將作業(yè)job提交至排產(chǎn)系統(tǒng)(scheduling system)。每一個作業(yè)由一系列任務集合組成(Each jobconsists of a set oftasks),這些作業(yè)的每一個都由排產(chǎn)系統(tǒng)通過映射放于一個集群的可用機器集合中。3.1 MapReduce鳥瞰 map的調(diào)用是通過自動將輸入數(shù)據(jù)分區(qū)到M分裂(M splits) 集合，從而得以分布在多臺機器之上。分裂后的輸入可由多臺機器并行運算。Reduce的調(diào)用是通過將中間鍵值對分區(qū)到R片(R pieces)來分散處理的，這種R片分區(qū)功能的例子如 hash(key) mod

13、R。 圖一顯示了MapReduce 操作的全局概覽（我們所實現(xiàn)的MapReduce接口方式）。 當用戶程序調(diào)用MapReduce 時， 以下序列的動作依次發(fā)生（圖一中的 數(shù)字記號與之對應）:（1）用戶程序的MapReduce 庫首先將輸入文件分割成M個小片，典型的小片大小從16M到64M不等（由用戶通過參數(shù)進行行配置。然后在一集群上開啟許多個分割程序拷貝。（2）上述的多個分割程序中的一個稱之為 master. 其余的拷貝由Master賦予work. 有稱之為M的 map任務和稱之為R的Reduce任務。master將喚醒空閑狀態(tài)的work線程，并賦予其map任務或Reduce任務。（3）被賦予

14、了map任務的工作線程讀取相應的輸入分割文件的內(nèi)容，改工作線程解析輸入文件的鍵值對并將其結(jié)果傳送到用戶在map功能中定義的鍵值對之中。有map功能產(chǎn)生的中間鍵值對被緩存在內(nèi)存當中。（4）被緩存的鍵值對周期性的寫到本地磁盤之上， 由分區(qū)程序分塊到R 個區(qū)域。在本地磁盤上緩存著的鍵值對的地址被傳遞回master, 這個master主管著將上述地址傳遞給reduce工作線程。（5）一旦reduce工作線程被master通知到這些地址，它就用遠程程序調(diào)用(remote procedure calls)讀取本地磁盤上的map工作線程產(chǎn)生的緩存數(shù)據(jù)。一旦reduce工作線程讀取所有的中間數(shù)據(jù)， 它就對這些

15、中間鍵值對按照鍵進行排序，一個自然的結(jié)果就是所有相同的group在一起了。排序是必要的，因為許多不同的鍵將映射到相同的reduce任務。如果這種排序?qū)τ趦?nèi)存中排序過大的話，就采用外部排序。（6）reduce 線程對經(jīng)過排序的中間數(shù)據(jù)進行迭代遍歷，將其遇到的每一個唯一中間鍵傳送給用戶定義的reduce功能，附帶傳送該鍵所對應的值的集合。（7）當所有的map任務和reduce任務全部結(jié)束， master 將喚醒用戶程序。就在此時，用戶程序中的“MapReduce調(diào)用”返回用戶代碼。 在成功的結(jié)束后， mapreduce的輸出存在于R 輸出文件中（每個reduce任務對應一個輸出R文件，文件名由用戶

16、定義）。典型情況下， 用戶無需將R輸出文件合并為一個文件， 他們通常將輸出文件再次作為另一個MapReduce 調(diào)用的輸入文件，或者從另一個“可以處理多個文件進行分割后的輸入”的分布式應用中進行使用。3.2 Master的數(shù)據(jù)結(jié)構(gòu) master保持了好幾個數(shù)據(jù)結(jié)構(gòu)。 對每一個 map任務以及reduce任務， master負責存儲其狀態(tài)（空閑，工作中或者完成）同時存儲非空閑任務的機器ID.master可以理解為這樣一種管道，通過它中間文件的區(qū)域位值就可以由map 任務傳送到reduce任務。因此,對每一個已經(jīng)完成了的map任務而言，master存儲了由Map任務產(chǎn)生的中間文件R的地址和大小。最

17、新更新后的大小位值信息在map任務結(jié)束的時候被收到。信息被以遞增的增量方式推送到進行中的reduce任務。3.3容錯 Fault Tolerance 由于MapReduce 庫被設計用于幫助跑在數(shù)百上千臺機器上的處理非常大數(shù)量級數(shù)據(jù)，就必須設計一套精致的容錯處理機制。工作線程失效 Master 周期性的ping每一個工作線程。如果在一給定時限內(nèi)沒有接收到工作線程的反饋， master就把該工作線程標記為失效。工作線程所完成的任意map任務將被重置為初始空閑狀態(tài)。 因此當對其它工作線程進行排產(chǎn)的時候可以被揀選到。相似的，任何進行中的map任務或者reduce任務如果其實施任務的工作線程宕掉了，則

18、這些任務亦需重置為空閑狀態(tài)，使被排產(chǎn)揀選稱為可能。如果遇到失效failure, 則完成了的map任務必須再一次的執(zhí)行，那是因為它們的產(chǎn)出是在宕掉的機器的本地磁盤中，顯見不可獲取。同樣遇到失效，完成了的reduce任務則無需再次執(zhí)行，那是因為Reduce的產(chǎn)出是放置與全局文件系統(tǒng)之中。當一個map任務第一次由工作線程A執(zhí)行，設若A由于某種原因宕掉了，這個map任務將由工作線程B繼續(xù)執(zhí)行，此時所有執(zhí)行reduce任務的工作線程被廣播“B接A執(zhí)行map這個任務”。任何未來得及從A讀取數(shù)據(jù)的reduce任務將從B中讀取數(shù)據(jù)。MapReduce 對大尺度范圍的工作線程失效情況有很強的容錯反彈能力。比如，

19、 在某一MapReduce 操作過程中，在一個集群上運行的網(wǎng)絡在某幾分鐘有80臺機器宕機，MapReduce 的master監(jiān)控線程僅僅是簡單的再一次執(zhí)行那些宕掉的機器先前處理的任務，繼續(xù)推進進度， 以最終完成MapReduce 操作。Master失效 讓master監(jiān)控線程周期性的設置上文所述的master數(shù)據(jù)結(jié)構(gòu)檢查點是容易的。如果master任務宕掉了，一份新的拷貝將從上一個檢查點狀態(tài)啟動。然而， 來考慮這樣一個場景：僅僅有單個master。這種情況下的宕機并不有趣：因此我們目前的實現(xiàn)方式是一旦master宕掉的話，就將MapReduce abort掉?？蛻羧绻矚g的話也可以檢查狀態(tài)以及

20、重啟MapReduce 操作。“表達失效”的語法 當用戶定義的map以及reduce 操作對其輸入值而言是精準的函數(shù)關系時，我們的分布式實現(xiàn)看起來就像是在一個不會出錯的順序執(zhí)行基礎上的進行產(chǎn)出的一整個程序 。我們依賴于對map而言最小單位的commits且reduce任務也在這種單位粒度上進行產(chǎn)出。每一個進行中的任務將其產(chǎn)出寫到一個私有的零時文件。reduce任務產(chǎn)生這樣一個零時文件，且一個map任務產(chǎn)出類似的R文件（一個R文件對應一個reduce任務）。當一個map任務完成的時候， 工作線程把一個包含R零時文件信息的消息傳送給master。如果master再次收到了已經(jīng)完成的map 任務消息

21、時，master將屏蔽掉這樣的冗余消息。否則 master把R文件的名字記錄到master的數(shù)據(jù)結(jié)構(gòu)。當reduce任務結(jié)束后， reduce工作線程自動的對零時輸出文件重新命名為最終輸出文件名，如果相同的reduce 任務在多臺機器上執(zhí)行的話，多重命名調(diào)用(multiple rename calls)將其聲明為同一個最終輸出文件。 我們依賴于原子化的重新命名操作(由linux文件系統(tǒng)提供)來保證最終的文件系統(tǒng)的狀態(tài)僅包含了reduce任務的單一機器執(zhí)行所產(chǎn)出的結(jié)果。 我們絕大多數(shù)的map和reduce操作都是確定的，且由于我們的語法等同于順序執(zhí)行這樣一個實事，使得程序員非常容易的推斷其程序的

22、行為正確與否。當map 和/或 reduce操作不是確定的時候，我們?nèi)耘f提供了雖然弱一點的但是可靠的語法。在提供非確定操作符情況下，某特定的reduce任務R1的產(chǎn)出等同于由非確定程序的順序執(zhí)行產(chǎn)生的R1。無論如何However, 某不同的reduce任務R2的產(chǎn)出對相應的“非確定程序不同順序執(zhí)行”R2產(chǎn)出是一致的。 考慮map任務M以及reduce任務R1與R2, 約定 e(Ri) (i = 1 or 2)是對被遞交了的Ri的執(zhí)行（確實有這樣的一個執(zhí)行）。更弱的語法會浮出水面是源于以下一種可能情形：因為e(R1)可能已經(jīng)讀取了由執(zhí)行M而產(chǎn)生的輸出，以及e(R2)可能已經(jīng)讀取M的一個不同的執(zhí)行

23、。3.4 文件存儲位置 網(wǎng)絡帶寬在我們的計算環(huán)境中是相對稀缺的資源。我們利用這樣一個實事來保有網(wǎng)絡的帶寬：由HYPERLINK /wiki/GlusterFSGFS所管理的輸入數(shù)據(jù)存儲在組成我們集群的那些機器的本地磁盤之上。GFS將每個文件分割為64M大小的塊，同時存儲大約3份分割出來的塊的拷貝，放于不同的機器之上。MapReduce 監(jiān)控將輸入文件的位值信息放入一賬簿，并試圖對一包含相應輸入數(shù)據(jù)副本的機器上之map任務進行排產(chǎn)。如果試圖排產(chǎn)未能實現(xiàn)，則將嘗試對map任務的輸入數(shù)據(jù)的就近副本進行排產(chǎn)（例如， 在同一聚群中的作為含有數(shù)據(jù)的宕掉機器的備份back機器中排產(chǎn)）。當在集群的顯要工作線程

24、部分運行大型MapReduce 操作時，大多數(shù)的輸入數(shù)據(jù)都是本地讀取且不耗用網(wǎng)絡帶寬。3.5 任務粒度 如上文所述，我們將map階段細分為M個小片且將reduce階段細分為R個小片。理想的情況下，M 和 R 的數(shù)量要大大多于工作中的機器數(shù)量。讓每一個工作機器執(zhí)行許多不同任務將提升動態(tài)負載平衡的效能，以及當某一機器宕掉時加速恢復的速度：某臺機器 所完成的諸多map任務可以分發(fā)傳送到集群中其它所有活躍的機器之上繼續(xù)工作。在我們的實現(xiàn)中對M以及R可以取的數(shù)值有實際的上下界限制，這是因為監(jiān)控線程master必須知曉M加上R的排產(chǎn)決策之時間復雜度。同時，保持在內(nèi)存中保持M乘R的時間復雜度（一個雖然微小但

25、卻恒定的影響內(nèi)存使用的因素是：O(M*R)片狀態(tài)幾乎總是由一map/reduce任務對一比特組成的）。 由于每個reduce任務以不同的輸出文件為結(jié)尾，所以R 通常受到來自用戶的約束。在實踐上， 我們傾向與選擇M 從而每個獨立的任務的輸入數(shù)據(jù)大約在16到64兆之間（這樣使得上述的位值優(yōu)化為最佳），并且我們將R設置為期望工作的機器數(shù)量的倍數(shù)。通常我們是以2千臺機器運行M為20萬，R為5千作為MapReduce計算的配置。3.6備份任務 一個延長MapReduce 操作總時間的常見現(xiàn)象是：一臺機器以異常長的時間來完成最后某幾個map任務或reduce任務中的一個。 這種異常長時間的現(xiàn)象可能在整個主

26、機范圍內(nèi)以多重原因浮現(xiàn)。比如， 一臺壞磁道的機器可能由于高頻糾錯而使讀取性能從30mb/s降低到1mb/s. 集群的排產(chǎn)系統(tǒng)可能已近對其它任務在某機器上排產(chǎn)好，從而導致由于CPU、內(nèi)存、本地磁盤或網(wǎng)絡帶寬資源的爭用而更慢的MapReduce 代碼執(zhí)行。 我們遇到過的一個問題是這樣一個bug, 初始化代碼導致處理器高速緩存功能關閉：被感染的機器將以低于原先百分之一的下降性能進行算。 我們用一種通用機制來消除上述性能下降現(xiàn)象。當一個MapReduce 操作接近完成，master對處理中的剩余任務之備份進行排產(chǎn)。無論原先的執(zhí)行還是備份的執(zhí)行完成時，任務都會被標記為完成。我們對這種機制進行了調(diào)優(yōu)從而增

27、加了耗用比不高于幾個百分點的計算資源。我們發(fā)現(xiàn)這將顯著的減少完成大型MapReduce 操作所耗用的時間。例子如5.3節(jié)所述排序，在關閉備份任務機制的情況下將延長44的時間以完成排序。4.精益求精 盡管簡單的Map和Reduce函數(shù)完全可以大多數(shù)滿足基本功能需求，我們發(fā)現(xiàn)作某些擴展仍然是有用的。4.1分區(qū)功能 MapReduce 的用戶指定了reduce任務數(shù)與輸出文件的期待數(shù)R. 數(shù)據(jù)經(jīng)由這樣一種任務“用到中間鍵的分區(qū)函數(shù)”被分區(qū)切割。一種默認的分區(qū)切割函數(shù)是由一哈希函數(shù)提供的，如hash(key) mod R 這傾向于生成相當平衡的分區(qū)。然而在其它的一些情況下， 用改建的另外一些函數(shù)來分區(qū)

28、也是有用的。比如,有時候輸出鍵是URL，我們想要所有的僅有單一宿主機的條目存儲到同一個輸出文件。為了支持這樣的情況， MapReduce 庫的用戶可以提供一特殊的分區(qū)切割函數(shù)。比如，把 hash(Hostname(urlkey) mod R 作為分區(qū)切割函數(shù)，使得所有來自于同一個主機的URL得以歸屬到同一輸出文件。4.2 確保順序 Ordering Guarantees 我們確保在一給定分區(qū)中，中間鍵值對以遞增的鍵順序進行處理。這種順序保證了，就每個分割而言，更容易的產(chǎn)出排過序的輸出文件。當輸出文件的格式需要支持“由鍵作為索引的高效隨機存取”這一情況時就有用了，或者輸出文件的用戶發(fā)現(xiàn)數(shù)據(jù)已排好

29、序是中良好的用戶體驗之時。4.3組合器Combiner函數(shù) 在有些情況下， 由每個map任務產(chǎn)生的中建鍵存有相當數(shù)量的副本，且用戶定義的Reduce函數(shù)展現(xiàn)出一種連續(xù)的和可交換的方式。2.1節(jié)所述單詞統(tǒng)計就是這種情況的一個例子。由于詞頻統(tǒng)計傾向于HYPERLINK /wiki/%E9%BD%8A%E5%A4%AB%E5%AE%9A%E5%BE%8BZipfHYPERLINK /wiki/%E9%BD%8A%E5%A4%AB%E5%AE%9A%E5%BE%8B齊夫分布，每一個map任務會產(chǎn)生成百上千條記錄，格式形如。所有這些記錄都會經(jīng)由網(wǎng)絡發(fā)送到唯一reduce任務，然后由Reduce功能加總起

30、來產(chǎn)生一個數(shù)值。我們允許用戶指定一可選組合器函數(shù)來局部歸并數(shù)據(jù)于網(wǎng)絡發(fā)送之前。每一執(zhí)行了map任務的機器都執(zhí)行了組合器函數(shù)。典型的，實現(xiàn)組合器與reduce 功能的代碼是同一份。兩者的唯一不同在于MapReduce 庫如何處理生成數(shù)據(jù)。reduce功能的輸出是寫到一個最終輸出文件的。一組合器函數(shù)的輸出是寫到一用于傳送止reduce任務的中間文件。局部歸并顯著的提升了MapReduce 操作中的一些類。 附錄A包含了用到組合器的例子。4.4 輸入輸出類型 MapReduce 庫對支持讀取不同格式的輸入數(shù)據(jù)類型。比如：文本模式的輸入將每一行視為鍵值對：鍵是文件中的位移，值是該行的內(nèi)容。另外一種常用

31、的支持格式存貯按鍵排序的鍵值對序列。每種輸入類型的實現(xiàn)均知道：在以單獨map任務進行處理時，如何將自身分割為有意義的區(qū)間（例如，文本模式的區(qū)間分割保證了區(qū)間分割僅發(fā)生在行邊界 range splits occur only at line boundaries）。雖然大多數(shù)用戶僅用了預定義類型中的一小部分，用戶仍然可以通過提供簡單閱讀器界面 （simple reader interface）的實現(xiàn)以支持新的類型。閱讀器不必必須提供從文件讀取的數(shù)據(jù)。比如很容易定義從數(shù)據(jù)庫中讀取記錄的閱讀器，或者從映射到內(nèi)存中的數(shù)據(jù)結(jié)構(gòu)的閱讀器。同樣的， 我們支持一輸出類型集用以由不同格式產(chǎn)生數(shù)據(jù)。同時對用戶代碼

32、來說，支持新輸出類型是同樣簡單的。4.5 邊際效應 在某些情況下，MapReduce 的用戶發(fā)現(xiàn)作為從map和/或reduce操作而產(chǎn)生的輸出文件的附帶，產(chǎn)出輔助文件是有益的。我們依賴于應用編寫器 (pplication writer )使這些邊際效應原子化，以及擁有冪等(idempotent)效應。（譯注：早期動態(tài)web架構(gòu)的重要特性就是冪等idempotent除非底層資源發(fā)生變化，否則同一請求的結(jié)果總是相同的。這意味著瀏覽器或代理服務器都可以在本地對 文檔進行緩存，只要底層資源沒有發(fā)生變化，那就可以從本地緩存中檢索資源，而不再需要從遠程服務器檢索。這種方法能提高用戶感受到的響應性，并增加系

33、統(tǒng)整體效率和可伸縮性。） 典型的，應用編寫器寫到一零時文件且一旦當這個文件完全生成時，自動的重命名該文件。我們不提供由單個任務產(chǎn)出的多重輸出文件的原子化二階承諾（ atomic two-phase commits）。因此，那些產(chǎn)出多重輸出文件的且有交叉文件一致性約束的任務應當是確定的。在實踐中這種約束卻沒有過。4.6屏蔽無效記錄 （Skipping Bad Records） 有時用戶代碼存在bug, 這肯定會引發(fā)Map或Reduce函數(shù)會在特定的記錄上crash。 類似的bug使得MapReduce 操作無法百分百完。通常情況下就是修復這個bug, 但有時候這種修復在源代碼級別是不可行的，因為

34、這些bug是由源代碼所不能影響到的第三方庫所引發(fā)。有時丟棄一小部分的記錄也是可行的，比如對某一大數(shù)據(jù)集合做統(tǒng)計分析的時候。當MapReduce 偵測到某條記錄必定會引發(fā)crash，就跳過這些無效紀律以使得任務可以繼續(xù)下去。 上述情況下我們提供了可選擇的執(zhí)行模式。每一個工作進程( worker process )都安裝“捕獲語法錯誤以及總線錯誤的”信號句柄。在喚醒用戶的Map 或Reduce操作之前，MapReduce在全局變量中存儲參數(shù)的序列值。如果是用戶代碼產(chǎn)生了信號， 信號句柄將傳送包含了序列值的微小(last gasp) UDP 包至MapReduce 監(jiān)控線程。當master已偵測到某

35、特定記錄上多于一次的失效，以及在mater對相應的Map或Reduce任務再次執(zhí)行時，會提醒這些記錄將被屏蔽。4.7本地執(zhí)行 Map或Reduce功能中的Debugging問題可以表現(xiàn)的很有技巧性，因為實際的計算是在數(shù)千臺機器組成的分布式系統(tǒng)范圍上，由master隨機指定工作分派的。 為了幫助查找bug，進行性能分析(profiling)以及小范圍測試，我們已開發(fā)出了MapReduce庫的另一實現(xiàn)版本，該版本會對某本地機器上的MapReduce操作順序執(zhí)行所有工作。控制有用戶提供，從而使計算被限定在特定的map任務。用戶通過特殊的標記來喚醒其程序，也可以容易的運用任何調(diào)式與測試工具（如 gdb

36、）。4.8狀態(tài)信息 監(jiān)控線程運行內(nèi)部HTTP服務器且輸出用戶可讀的狀態(tài)頁信息。狀態(tài)頁顯示了計算的進度， 比如多少任務已經(jīng)完成了、還有多少在進行中、 輸入的比特數(shù)、 中間數(shù)據(jù)的比特數(shù)、 輸出的比特數(shù)以及處理比率，等等。狀態(tài)頁同時提供了到標準錯誤以及經(jīng)由每個任務產(chǎn)出的標準輸出文件的鏈接。用戶可以用這種數(shù)據(jù)來預測計算大概會耗時多少，以及在以后的計算中是否將需要更多的外部資源。用戶頁同時可用于當計算比預期慢了許多的情況下進行推斷。 頂層狀態(tài)頁還顯示了哪臺工作機器宕掉了，以及它們所處理的那些處于進行中的map和reduce任務。這些信息對于試圖診斷用戶代碼中的bugs是有效的。4.9計數(shù)器 MapRed

37、uce 庫提供了計數(shù)器功能以統(tǒng)計各種事件出現(xiàn)的次數(shù)。比如，用戶代碼可能要統(tǒng)計處理單詞的總數(shù)，或者諸如已索引了的德語文件數(shù)這樣的統(tǒng)計。為這種用這種計數(shù)器，用戶代碼可以先生成命名計數(shù)器對象，然后適量的在Map和/或Reduce功能中增加這種命名計數(shù)器對象。比如：Counter* uppercase;uppercase = GetCounter(uppercase);map(String name, String contents): for each word w in contents: if (IsCapitalized(w): uppercase-Increment(); EmitInter

38、mediate(w, 1);從獨立工作機收集來的計數(shù)器值周期性的傳遞至master( 背負從ping 而來的響應Piggybacked on the ping response ) 監(jiān)控線程加和來自正常的(successful)map和reduce任務的計數(shù)器值, 并在MapReduce 操作完成時將計數(shù)值返回給用戶代碼。當前的計數(shù)器值也在主機狀態(tài)頁中顯示，這樣就可以使人實時的監(jiān)控動態(tài)計算的進度。當增加計數(shù)器值的時候， master消除同時執(zhí)行相同map或reduce任務所導致的重復冗余計算。（重復冗余計算這種情況的發(fā)生是由于使用了備份任務，或者在失敗時重復執(zhí)行所產(chǎn)生的）。 諸如已經(jīng)處理的輸入

39、鍵值對的數(shù)量以及已近處理了的輸出鍵值對的數(shù)量這樣一些計數(shù)器的值是由MapReduce 庫自動維護的。 比如在有些MapReduce 操作中，用戶代碼可能需要確保輸出鍵值對的數(shù)目與處理過的輸入鍵值對的數(shù)目精確相等，或者處理過的德語文件的那部分屬于某個已處理文件總數(shù)的可容部分（tolerable fraction）。5.性能 在本節(jié)中我們將用運行于大集群上的兩個計算來度量MapReduce的性能。第一個計算在1TB的數(shù)據(jù)中搜索特定的模式。第二個計算代表了用戶定義的MapReduce的一大子集 ，該子集的一個類將數(shù)據(jù)從一種格式表達為另外一種格式；另外一個類負責從一個大的數(shù)據(jù)集合中萃取少量感興趣的數(shù)據(jù)

40、。5.1集群配置 所有的程序都是在約1800臺機器組成的集群之上允許的。每臺機器配置有2G主頻且支持超序執(zhí)行（HYPERLINK /wiki/Hyper-threadingHyper-Threading）的英特爾至強處理器（Xeon processors），兩臺160G的IDE硬盤以及千兆以太網(wǎng)鏈接。機器布置在這樣的網(wǎng)絡環(huán)境中：根節(jié)點總帶寬為100200Gbps且擁有二個層級的樹形交換網(wǎng)。所有機器處于相同的主機設施 （hosting facility）中。所以在任意機器對之間的一個來回，時間耗用少于百萬分至一秒。 在整個4GB的內(nèi)存中， 大約11.5GB保留給運行于該集群之上的其它任務。該程序

41、于一個周末運行，其時CPU, 硬盤以及網(wǎng)絡都處于空閑狀態(tài)。5.2 Grep Grep 命令行程序通過掃描10的10次方個100比特記錄，查找相對稀缺的三字母模式（該模式發(fā)生于92,337條記錄）輸入數(shù)據(jù)被分割成64兆的小片(M = 15000)，整個輸出放于一個輸出文件（R = 1）。圖二顯示了計算隨時間流逝的相應進度。 Y軸顯示了隨掃描到的輸入數(shù)據(jù)而變化的速率。隨更多機器參與到MapReduce 計算中來，速率也漸漸的提升起來，在1764臺機器參與到計算時達到了其峰值30GB每秒。當map 任務結(jié)束時， 速率開始向下回落并在80秒后返0.整個計算過程從開始帶結(jié)束大約耗費150S。這還包含了大

42、概1分鐘的常用（overhead）啟動時間。該常用時間是由“將程序傳遍布所有工作機器”決定的。并延遲與集群文件系統(tǒng)GFS 的交互以打開1000個輸入文件并為本地優(yōu)化（locality optimization）獲取信息。5.3排序 排序程序?qū)?0的10次方個100字節(jié)的記錄進行排序（相當于1TB數(shù)據(jù)）該程序是模仿基準兆排序程序（TeraSort benchmark）。排序程序由少于50行的用戶代碼組成。一個三行的Map功能從文本行中萃取出10比特的排序關鍵字，并且將關鍵字和初始文本行作為中間鍵值對。我們用一內(nèi)建的身份認真函數(shù)（built-in Identity function）作為Reduc

43、e操作。該函數(shù)將中間鍵值對不做任何修改就作為輸出鍵值對進行傳遞。最終排序之后的結(jié)果寫到2路復合GFS文件 （2-way replicated GFS files）。 和以前一樣， 輸入數(shù)據(jù)同樣被分割成了64M的小片（M = 15000）。我們將排好序的輸出放入4000個文件（R = 4000）。分割程序用鍵的初始字節(jié)將這些排好序的輸出分別放入R個片中的每一片。 對這個基準來說，我們的分區(qū)程序有內(nèi)建的關于鍵分布的相關知識。就通用排序程序而言，我們會添加一個“會收集鍵的樣本的”預處理MapReduce操作，并且用這個樣本鍵的分布來計算出最終排序過程的分裂點 （compute splitpoints

44、 for the nal sorting pass）。圖三（a）顯示了排序程序的一次通常執(zhí)行的進度。左上圖顯示了讀取輸入的速率。速率在13GB/s時達到峰值并由于所有的map任務結(jié)束于200秒內(nèi)，故回落的相當快速。注意到輸入速率要慢于grep，這是由以下原因所造成的：排序map 任務花費了它們一半的時間，且I/O帶寬是立即將輸出寫到了它們的本地磁盤之中。grep的相應中間輸出可以忽略其大小。 左邊中間的圖顯示了數(shù)據(jù)經(jīng)由網(wǎng)絡從map任務傳輸?shù)絩educe任務的速率。一旦第一個map任務完結(jié)時，shuffling就開始了。 圖形中的第一個隆起是對將近1700個reduce任務的備份執(zhí)行所致。（整個

45、MapReduce 被指派到約1700臺機器， 且每臺機器一次最多執(zhí)行一個reduce任務）大約300秒的時間被花在計算上， 第一批batch的reduce任務中的某些完成后我們對剩余reduce任務開始慢慢移動數(shù)據(jù)（shuffling data）。所有的移動（shuffling ）在600秒內(nèi)處理完進入計算。 左下底部圖形顯示了這樣的速率：排序后的數(shù)據(jù)被reduce任務寫到最終輸出文件。由于機器忙于排序中間數(shù)據(jù)，因此在“第一個慢慢移動shuffling時間段的結(jié)束點和寫時間段的開始”之間產(chǎn)生了一個延遲。寫的速率維持在24GB每秒左右。所有寫結(jié)束于850秒內(nèi)進入計算。包含了啟動后的正常費時，整

46、個計算花費了891秒。這個結(jié)果和目前最好的報告結(jié)果（就TeraSort benchmark耗費 1057 秒而言）是相似的。【文見：HYPERLINK /cs/spsort.pdf如何快速的對TB數(shù)據(jù)排序】【18】 一些值得關注的現(xiàn)象是：輸入速率高于移動速率以及輸出速率。那是因為我們做了以下優(yōu)化：大多數(shù)數(shù)據(jù)都是從本地磁盤讀取以繞開相對受限的網(wǎng)絡帶寬。移動速率較輸出速率為快，因為在輸出階段要寫輸出數(shù)據(jù)的兩份拷貝（該備份策略是出于可靠性以及可用性考慮 reliability and availability）。由于我們的nix系統(tǒng)所提供的可靠性以及可用性機制，所以我們寫了兩個備份。為寫數(shù)據(jù)而必須的

47、網(wǎng)絡帶寬在文件系統(tǒng)使用抹除碼（HYPERLINK /wiki/Erasure_codingerasure coding）而非復制的情況下將會下降。5.4備份任務之效應 在圖3(b)中，我們顯示了排序的一次無備份任務的執(zhí)行。執(zhí)行流與圖3(a)是相似的，除了一點：圖3(b)有一個非常長的尾(Done)以至很難有寫活動發(fā)生。在960秒以后，除了5條reduce任務其余都結(jié)束了。這些少數(shù)的reduce任務直到300秒之后仍未結(jié)束。整個計算1283秒，逝去時間增加了44% （an increase of 44% in elapsed time）。5.5 機器失效 （Machine Failures） 在

48、圖3(c)中，我們展示了計算中的某幾分鐘內(nèi)故意關閉掉1764臺工作機器中的200臺后的排序執(zhí)行情況。集群調(diào)度程序立即在這些機器上啟動了新的工作進（如果僅僅是進程被kill掉，那么機器仍運行無礙）。 由于相應的map工作線程被kill掉，導致先前已完成的map工作消失掉，這就需要重新執(zhí)行map任務。所以工作進程的失效可以被認為是一種消極的輸入。 這種map工作的再執(zhí)行發(fā)生的相對快速（happens relatively quickly）。整個計算包含了通常的啟動在933秒內(nèi)結(jié)束（僅僅比圖3(a)通常執(zhí)行時間要多5%）。6.經(jīng)歷 我們在2003年初寫出了第一個Experience庫版本，并在200

49、3年八月對其做了重大改進，包含了本地優(yōu)化，工作主機上執(zhí)行任務的動態(tài)負載平衡。從那以后我們很驚訝同時樂見其成的發(fā)現(xiàn)：MapReduce庫的應用范圍就我們手頭的那類問題來說是多么廣泛。它廣泛的運用于谷歌中多處問題域：大規(guī)模機器學習問題谷歌新聞的集群問題萃取數(shù)據(jù)用于產(chǎn)生報告新實驗或新產(chǎn)品的網(wǎng)頁頁面萃取數(shù)據(jù)（比如從“大規(guī)模本地化搜索網(wǎng)頁的”語料庫中萃取地理位域），以及大比例圖計算 圖4顯示了獨立的MapReduce程序的個數(shù)在我們的源代碼管理系統(tǒng)中，隨時間增長的顯著增加趨勢。 MapReduce 如此成功是因為：該模型使“寫一個簡單的程序，并使其有在半個小時期間內(nèi)效的運行于數(shù)千臺機器”成為可能；大大加

50、速了開發(fā)以及原型設計周期 ( prototyping cycle )。對程序員更加有用的是該模型允許在毫無分布式或并行系統(tǒng)經(jīng)驗儲備前提下，開發(fā)大規(guī)模數(shù)量的資源成為可能 （it allows programmers who have no experience with distributed and/or parallel systems to exploit large amounts of resources easily）。在每個job結(jié)束時，MapReduce 庫會對該job使用了的計算資源作日志統(tǒng)計。表1， 我們顯示了2004年八月運行于Google上的MapReduce job子集

51、的一些統(tǒng)計。6.1 大尺度索引 到目前為止，在MapReduce 的所有重要應用中，我們已近完全重寫了索引系統(tǒng)的產(chǎn)品，該產(chǎn)品產(chǎn)生“用于谷歌網(wǎng)頁服務搜索的”數(shù)據(jù)結(jié)構(gòu)。索引系統(tǒng)將輸入視為我們的伺機而動的系統(tǒng)（crawling system）已經(jīng)收到的文檔的一個大的集合，其后將之存儲為GFS 文件。這些文件的原始內(nèi)容數(shù)據(jù)都多于20TB。索引過程通常耗費5到10個MapReduce操作組成的序列。使用MapReduce（較之用于索引系統(tǒng)的以前的HYPERLINK /wiki/Ad-hocad-hoc分布式處理版本）帶來了以下幾個好處：由于代碼擁有了容錯功能，以及分布以及并行特性被隱含到MapReduc

52、e 庫中，所以索引代碼更簡單、更小且易于理解。比如，計算的某一階段之大小從原先的C+代碼量約3800行降到了用MapReduce表達的700行。MapReduce 庫的性能優(yōu)異，使得我們能將概念上不相關的計算保持獨立，而不是將其混合起來以避免數(shù)據(jù)的冗余處理(avoid extra passes over the data)。比如，一個改變就是原先要花個把月的老版本索引系統(tǒng)現(xiàn)在只需數(shù)天。由于大多數(shù)由宕機，機器慢速以及網(wǎng)絡不穩(wěn)定（hiccups）等問題都自動由MapReduce庫無需外部介入解決了，所以索引過程變得非常容易操作。更重要的是：通過對索引集群（indexingcluster）增加新的機

53、器節(jié)點，提升索引過程的性能變得更加容易。比如，相連功能（associative function）可以這樣計算：在由N個元素組成的數(shù)組之前綴上，在logN的時間復雜度內(nèi)，在N個處理器上進行的并行前綴計算。7.相關工作 許多系統(tǒng)提供受限編程模型（restricted programmingmodels ）并使用該約束對計算并行自動化。比如，相連功能（associative function）可以這樣計算：在由N個元素組成的數(shù)組之前綴上，在logN的時間復雜度內(nèi)，在N個處理器上進行并行前綴計算【6，9，13】。MapReduce 模型可以被看作“基于我們關于真實世界計算的經(jīng)驗”(based on

54、our experiencewith large real-world computations）這類模型之簡化和提純。更重要的，我們在數(shù)千臺處理器的尺度上實現(xiàn)了容錯處理。而作為對比，大多數(shù)并行處理系統(tǒng)（parallel processing systems）僅僅是在小范圍內(nèi)作了實現(xiàn)并且是將機器失敗需要處理的細節(jié)交給程序員去把握。 巴爾克的同步計算【HYPERLINK /cites/Document/A-bridging-model-for-parallel-computation.htmlA bridging model for parallel computation】【17】以及一些HY

55、PERLINK /wiki/Message_Passing_InterfaceMPI原語【11】提供了更高級別的抽象【文見： Portable Parallel Programming with the Message-Passing Interface】，以使得程序員更容易寫出并行程序。這些系統(tǒng)和MapReduce 的一個關鍵區(qū)別就是MapReduce 利用了受限編程模型來自動化的并行用戶程序并以透明的方式提供了容錯處理機制。 我們本地優(yōu)化的靈感來自于類似活動桌面這樣的技術【12，15】。計算被推送到接近本地磁盤的處理單元，以減少數(shù)據(jù)在網(wǎng)絡或I/O子系統(tǒng)上的傳輸。我們運行在小數(shù)量的磁盤直接連

56、接的商用處理器上，而非直接運行在磁盤處理器（disk controller processors）上，總的方法是相似的。我們的備份任務機制相似于HYPERLINK /viewdoc/summary?doi=25.6087卡洛特系統(tǒng)中的排程機制【3】。簡單早期排程的一個缺陷是如果一個給定的任務會引發(fā)周期性的失敗，則整個計算就會失敗。我們通過忽略不可用記錄的方式修復了該問題的幾個實例。 MapReduce 的實現(xiàn)依賴于這樣的內(nèi)部集群管理系統(tǒng)：它負責在大規(guī)模共享機器上分發(fā)與運行用戶任務。雖然不是這篇paper的目的， 集群管理系統(tǒng)與諸如HYPERLINK /wiki/Condor_High-Thro

57、ughput_Computing_SystemCondor這樣的系統(tǒng)是非常相似的【16】。 MapReduce庫的排序功能與高性能網(wǎng)絡排序（HYPERLINK /wzhang/dbpapers/nowsort.pdfNOW-Sort）【1】上的操作是相似的。源機器（即 map 工作機）將數(shù)據(jù)分割排序后發(fā)送給R個reduce工作者中的一個。每個reduce工作者在本地排（如果可能的話放在內(nèi)存中）對數(shù)據(jù)排序。當然NOW-Sort沒有用戶定義的Map以及Reduce功能（這兩個功能使得我們的庫得以廣泛的應用）。 HYPERLINK /viewdoc/summary?doi=5.4077River【2

58、】提供了一種編程模型，進程得以通過在分布式隊列上發(fā)送數(shù)據(jù)來互相通信。類似MapReduce， River系統(tǒng)即便在“由異構(gòu)硬件或系統(tǒng)擾動所引進的非均勻性”（non-uniformities introduced by heterogeneous hardware or system perturbations）情況下，仍然試圖提供良好的平均用例性能。River是通過仔細規(guī)劃磁盤以及網(wǎng)絡傳輸來平衡各個結(jié)束時間。MapReduce卻有一個不同的方法。通過約束程序模型， MapReduce 框架可以將問題隔離到一個較大數(shù)目的且更細微粒度的任務上。這些任務是在可以獲取的工作者為條件下動態(tài)排程的，故而更

59、快的工作者處理更多的任務。受限編程模型同時也允許對接近job結(jié)束時的“任務冗余執(zhí)行”進行排程，這種冗余執(zhí)行極大的延緩了“以非均勻形式出現(xiàn)的 (in the presence of non-uniformities)，諸如慢的、停頓不前的工作者的”完成時間。 批處理分布式文件系統(tǒng)HYPERLINK /thain/library/badfs-nsdi04.pdfBADFS【5】相對MapReduce有一迥異的編程模型，該模型關注job在廣域網(wǎng)范圍的執(zhí)行.。但是兩個模型卻在兩點上基本一致：(1)兩者編程模型都使用備份冗余在宕機等引起數(shù)據(jù)丟失時找回數(shù)據(jù); (2)兩者都使用目標為止特性（locality

60、-aware）的排產(chǎn)以降低擁擠網(wǎng)絡連接時的數(shù)據(jù)發(fā)送量。 HYPERLINK /brewer/cs262b/TACC.pdfTACC【7】是一種基于集群的可伸縮網(wǎng)絡服務系統(tǒng)，設計用以簡化“高可用性網(wǎng)絡服務的”構(gòu)造。就像MapReduce， 它依賴于再次執(zhí)行機制以實現(xiàn)容錯。8.結(jié)論 MapReduce 程序模型已成功運用于谷歌的許多不同領域。我們認為這種應用的成功要歸功于以下幾個方面。首先由于該模型隱藏了并行、容錯、本地優(yōu)化以及負載平衡的細節(jié)，所以即便是那些沒有并行和分布式系統(tǒng)經(jīng)驗的程序員也易于使用該模型。其次MapReduce 計算可以很容易的表達大量的各種問題。比如，MapReduce 用于為