摘要：

MapReduce是一個編程模型，以及處理和生成大型數據集的一個相關實現，它適合各種各樣的現實任務。用戶指定計算的map和reduce函數。底層運行系統自動地將大規模集群機器間的計算并行化，處理機器故障，以及調度機器間通信以充分利用網絡和磁盤。程序員會發現這個系統很好使用：在過去的去年中，超過一萬個不同的MapReduce程序已經在Google內部實現，平均每天有十萬個MapReuce作業在Google集群上被執行，每天總共處理20PB以上的數據。

1 簡介

在MapReduce開發之前，作者和其他許多的Google員工實現了數以百計的處理大量原始數據(如抓取到的文檔、Web請求日志等等)的專用計算方法，以計算各種導出的數據，如倒排索引、Web文檔圖結構的各種表示、每個host抓取到的頁面數的總結、某一天最頻繁的一組查詢。大多數這樣的計算在概念上是非常簡單的，然而它們的輸入數據量通常非常大。為了在合理的時間內完成這些計算，它們必須分布到成百上千的機器上。如何并行化計算，分發數據，以及處理故障，這些問題結合起來，往往會讓程序員使用大量復雜代碼來處理，而掩蓋了原本簡單的計算。

為了應對這一復雜性，我們設計了一個新的抽象，它允許我們表達試圖執行的簡單計算，但將并行化、容錯、數據分布和負載均衡等凌亂的細節隱藏到了庫中。這個抽象的靈感來源于出現在Lisp和許多其他函數式語言中的map和reduce原語。我們實現了大部分的計算，包括為輸入的每一個邏輯記錄應用一個map操作以計算一組中間鍵值對，然后對所有共享同一個鍵的值應用一個reduce操作以恰當地結合導出的數據。此函數式模型支持用戶自定義map和reduce操作，使我們能非常容易地并行處理大型計算，和使用再執行(reexecution)作為主要的容錯機制。

這項工作的主要貢獻就是一個簡單而強大的接口，它完成自動并行化、大規模分布計算，結合該接口的一個實現在大型商用PC集群上獲得了很高的性能表現。該編程模型還可以用于同一臺機器上多個核心間的并行計算。

第2部分描述了基本的編程模型并給出幾個例子。第3部分描述了MapReduce接口專門針對基于集群的計算環境的一個實現。第4部分描述了我們發現的這個編程模型的幾個很有用的改進(refinements)。第5部分描述了對各種不同任務的實現的性能度量。第6部分探索了MapReduce在Google中的應用，包括使用它作為重寫我們的生產索引系統的基礎的一些經驗。第7部分討論了相關和未來的工作。

2 編程模型

這個計算需要一組輸入鍵/值對，并生成一組輸出鍵/值對。MapReduce庫的使用者將計算表達為兩個函數：map和reduce。

map，由用戶編寫，需要一對輸入并生成一組中間鍵/值對。MapReduce庫將所有與相同鍵值 I 相關聯的值組合到一起，并將它們傳遞給reduce函數。

Reduce函數，同樣由用戶編寫，接受中間鍵 I 和這個鍵的一組值。它將這些值合并以形成一組可能更小的值。通常每次reduce調用只生成0個或1個輸出值。中間值靠一個迭代器提供給用戶的reduce函數。這使我們能夠處理大量太大以至于不能裝入內存的值列表。

2.1 例子

考慮一下在一個巨大的文檔集合中統計每個單詞出現次數的問題。使用者會編寫與下面偽代碼類似的代碼：

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

map函數發出每個單詞加一個相關的出現次數(count)(在這個簡單例子中僅為1)。reduce函數對發給一個單詞的所有數(count)求和。

此外，用戶編寫代碼將輸入和輸出文件名以及可選的調優參數填入mapreduce規范對象中。然后調用MapReduce函數，將它傳遞給規范對象。用戶的代碼與MapReduce庫(C++實現)相連接。我們最初的MapReduce資料中有這個例子的完整程序【8】。

2.2 類型

盡管前面的偽代碼是按照輸入輸出字符串形式編寫的，概念上由用戶提供的map和reduce函數是有相關類型的。 

map (k1, v1) --> list(k2, v2)  
reduce (k2, list(v2)) --> list(v2) 

也就是說，輸入鍵和值與輸出鍵和值來自不同的域。此外，中間鍵和值與輸出鍵和值來自同一個域。

3 實現

MapRedue接口的許多不同實現都是可能的。正確的選擇取決于環境。例如，一種實現可能適合一個小型的共享內存的機器，另外一種可能適合一個大型的NUMA多處理器，而另外一種可能適合一個更大的聯網計算機集合。在我們最初的文章發表以后，已經發展出了很多MapReduce的開源實現【1， 2】，MapReduce在各種問題領域的適用性也得到了研究【7， 16】。

這一部分描述了我們的一種MapReduce實現，其目標是目前廣泛應用在Google中的計算環境：由交換千兆以太網連接在一起的大型PC集群【4】。在該環境中，機器通常運行Linux系統，有雙核 x86 處理器以及4-8GB內存。個別機器擁有1GB/s的網絡帶寬，但每臺機器等分的帶寬遠遠低于1GB/s。一個計算集群包含了成千上萬臺機器，因此機器故障是很常見的。存儲由直接附在單獨機器上的廉價IDE磁盤提供。GFS，Google內部開發的一個分布式文件系統【10】，用來管理存儲在這些磁盤上的數據。文件系統使用復制來提供不可靠的硬件之上的可用性與可靠性。

使用者提交 jobs 給調度系統。每個 job 包含一組任務，且由調度程序映射(mapped)到集群間的一組可用的機器上。

3.1 執行概述

通過自動將輸入數據分割為一個有M個分裂(splits)的組，map調用分布在多臺機器間。輸入分裂可以由不同的機器并行處理。reduce調用通過利用分割函數(比如，hash(key) mod R)將中間鍵空間劃分為R片進行分布。分割數R和分割函數都是由使用者指定的。

圖1展示了在我們的實現中MapReduce操作的整體流程。當用戶程序調用MapReduce函數，以下順序行為將會發生(圖1中標記的數字對應下面列中的數字)。

用戶程序中的MapReduce庫首先將輸入文件劃分為M片，通常每片16~64MB(由用戶通過可選參數控制)。然后啟動集群中程序的多個副本。

這些程序副本中有一個特殊的master副本。其他副本則是由master分配了work的workers。集群中需要分配 M 個 map tasks 和 R 個 reduce tasks。master挑選閑置的workers且為每個worker分配一個 map task 或 reduce task。

分配了 map task 的一個worker讀取相應輸入劃分的內容。它從輸入數據中解析出鍵/值對并將每一對傳遞給用戶定義的map函數。由map函數產生的中間鍵/值對緩沖在內存中。

緩沖區的鍵/值對定期地寫入本地磁盤，由partition函數劃分到 R 個區域中。這些本地磁盤上的緩沖對的位置被傳遞會master，它將負責轉發這些位置給 reduce workers。

當一個 reduce worker 被 master 通知了這些位置后，它使用遠程進程調用來讀取來自map workers的本地磁盤中的緩沖數據。當reduce worker讀取到了所有分區中的中間數據后，它按照中間鍵將其排序，從而使所有相同鍵的出現次數組合在了一起。排序是必要的，因為通常很多不同的鍵被map到了同一個reduce task。如果中間數據太大以至于不能放在內存中，還需要使用一個外部的排序。

reduce worker對排序好的中間數據執行迭代，對每個唯一的中間鍵，它將這個鍵以及相應的一組中間值傳遞個用戶的 reduce 函數。reduce 函數的輸出被附加到這個reduce分區的最終輸出文件中。

當所有的 map tasks 和 reduce tasks 都完成后，master喚醒用戶程序。在這一點上，用戶程序的MapReduce調用返回到用戶代碼處。 

execution

成功完成后，mapreduce執行的輸出可以在R個輸出文件中獲得(每個reduce task一個，由用戶指定文件名)。通常，用戶無需將這R個輸出文件合并到一個文件中;他們通常將這些文件作為另一個MapReduce調用的輸入，或者在來自另外一個可以處理劃分到了多個文件中的輸入的分布式應用程序中使用它們。

3.2 master數據結構

master中有多種數據結構。對每一個map task和reduce task，它存儲了其狀態信息(限制，進行，或完成)和worker機器的身份(對于非閑置tasks)。

master是map tasks傳播中間文件區域位置到reduce tasks的導管。因此，對于每個完成了的map task，master存儲由這個map task生成的R個中間文件區域的位置和大小。master在map tasks稱后接收到這些位置和大小信息的更新。這些信息將逐步推送到正在進行reduce tasks的workers中。

3.3 容錯

由于MapReduce庫旨在幫助利用成百上千的機器來處理大量數據，它必須優雅地容忍機器故障。

• 處理worker故障

master會定期地ping每一個worker。如果在一定時間內沒有收到來自某臺worker的響應，master將這個worker標記為故障。任何由worker完成的map tasks都被重置為初始閑置狀態，因而可以在其他的workers中調度。同樣，在故障worker上的任何正在進行的map task和reduce task也被重置為閑置狀態以便進行重新調度。

故障worker上已完成的map task需要重新執行，因為它們的輸出存儲在了故障機器的本地磁盤中導致無法訪問。已完成的reduce tasks無需重新執行，因為它們的輸出存儲在了全局文件系統中。

當一個map task首先由worker A執行然后又由worker B執行(因為A發生了故障)，所有執行reduce task的workers將被通知重新執行。任何還未從worker A讀取數據的reduce task將從worker B讀取數據。

MapReduce適應于大規模的worker故障。例如，在一個MapReduce操作中，在運行中的集群上的網絡維護導致了一組80臺機器在幾分鐘內無法到達。MapReduce master簡單地重新執行無法到達的worker機器的工作且繼續前進，最終完成MapReduce操作。

• 語義失敗

當用戶提供的map和reduce操作是它們他們的輸入值的特定函數時，我們的分布式實現生成的輸出將與整個程序的無錯順序執行生成的輸出相同。

我們依靠map和reduce任務輸出的原子的提交來實現這一性質。每個正在進行的task將其輸出寫入私有臨時文件中。一個reduce task生成一個這樣的文件，map task生成R個這樣的文件(每個reduce task一個)。當一個map task完成后，worker發送一條消息給master，這條消息中包含了R個臨時文件的名字。如果master接收到了來自一個已完成的map task的完成消息，它將忽略這條消息。否則，它將這R個文件名記錄到master數據結構中。

當一個reduce task完成后，reduce worker自動重命名其臨時輸出文件為最終輸出文件。如果同一個reduce task在多臺機器上執行，同一個最終輸出文件的多個重命名調用將被執行。我們依靠由底層文件系統提供的原子的重命名操作來保證最終文件系統狀態僅包含來自一個reduce任務執行生成的數據。

絕大多數的map和reduce操作是確定的，事實上，我們的語義等價于這種情況下的一次順序執行，這使得程序員能夠非常容易地推斷程序的行為。當 map 和/或 reduce 操作不確定時，我們提供了較弱但仍然合理的語義。在不確定操作存在時，一個特定reduce task R1的輸出等價于由非確定性程序的一次順序執行R1生成的輸出。然而，另一個不同的reduce task R2的輸出可能對應該非確定性程序的另一個不同順序執行R2的輸出。

考慮map task M和reduce task R1和R2。令 e(Ri) 作為作為R1的執行(這確實是一個這樣的執行)。較弱的語義出現因為 e(R1) 可能讀取了M的一次執行生成的輸出，e(R2)可能讀取了M的另一次執行生成的輸出。

3.4 局部性

在我們的計算環境中，網絡帶寬是一個相對稀缺的資源。我們靠充分利用輸入數據(由GFS管理【10】)存儲在組成集群的機器的本地磁盤中這一事實來節省網絡帶寬。GFS將每個文件分成64MB的塊且在不同機器上存儲了每個塊的多個副本(通常3個)。MapReeuce master考慮每個輸入文件的位置信息且試圖調度一臺含有相應輸入數據的機器上的一個map task。如果失敗，它將試圖調度與該任務的輸入的復制品相鄰的一個map task(例如，同一網絡交換機中包含相同數據的兩臺機器)。當在一個集群的 workers 重要部分運行大型MapReduce操作時，大多數輸入數據都是本地讀取的，并不消耗網絡帶寬。

3.5 Task粒度

我們將map階段細分為M個片段，reduce階段細分為R個片段，如前所述。理想情況下，M和R應該遠高于worker機器的數量。每個worker執行多個不同tasks改善了負載均衡，且當一個worker故障后加快了恢復速度：它完成的多個map tasks可以分布到所有其他worker機器上重新執行。

由于master必須做O(M+R)此調度決策和在內存中保持O(M*R)個狀態，如前所述，在我們的實現中M和R的數量大小是有實際界限的。(然而，內存的使用量很小。O(M*R)個狀態中大約包含每個map/reduce task對一字節的數據。)

此外，R通常受到用戶限制，因為每個reduce task的輸出最終保存在一個單獨的輸出文件中。在實踐中，我們傾向于選擇M因而每個獨立task大約有16MB到64MB的輸入數據(因而之前所述的局部優化達到最搞笑)，且我們讓R是我們希望使用的機器數量的一個小的倍數。我們通常以M=200000, R=5000，使用2000臺worker機器執行MapReduce。

3.6 備份Tasks

延長MapReduce操作總時間的一個普遍原因是一個掉隊者(straggler)，也就是說，在這個計算中有一臺機器花了異常長的時間來完成最后幾個map或reduce tasks。掉隊者會以一大堆的理由出現。比如說，一臺擁有壞磁盤的機器可能經歷頻繁的矯正錯誤從而使讀取性能從30MB/s降低到了1MB/s。集群調度系統可能在這個機器上調度了其他任務，導致它更慢地執行MapReduce代碼，由于競爭CPU、內存、本地磁盤或網絡帶寬等資源。我們經歷的一個最近的問題是機器初始化代碼中的一個bug導致處理器緩存失效：受影響的機器計算速度放慢了100倍。

我們有一個通用機制來減輕掉隊者問題。當一個MapReduce操作接近完成時，master將調度還在進行的任務的備份執行。無論是原始或者備份執行完成，這個任務都被標記為完成。我們調整了這個機制，因而它增加了該計算的計算資源的使用，但不超過幾個百分點。我們發現它大大降低了完成大型MapReduce操作的時間。作為一個例子，當沒有備份task機制時，在5.3部分描述的排序程序多花了44%的時間完成。

4 改進

雖然由簡單編寫的map和reduce函數提供的基本功能已足以滿足大多數需求，我們發現了一些有用的擴展。這包括：

• 用戶指定的分區(partition)函數來決定如何將中間鍵值對映射到R個reduce碎片;
• 排序保證：我們的實現保證這R個reduce分區中的每個，中間鍵值對都按鍵的升序處理;
• 用戶指定的結合(combiner)函數的作用是，在同一個map task內，對按照同一個鍵生成的中間值進行局部結合，以減少必須在網絡間傳輸的中間數據數量;
• 自定義輸入輸出類型，為了讀新的輸入格式和生成新的輸出格式;
• 在單機上執行簡單debug和小規模測試的一種方式。
• 在【8】中有對這幾項的詳細討論。

5 性能表現

在此部分，我們利用大型集群上的兩個計算來測量MapReduce的性能表現。一個計算通過搜索大約1TB的數據來找到一個特定的模式。另一個計算對大約1TB的數據進行排序。這兩個程序代表由MapReduce用戶編寫的真正程序的一個大的子集-----程序的一個類用來從一個表示(representation)向另一個表示shuffle數據，另一個類從大數據集中提取小部分關注的數據。

5.1 集群配置

所有程序都在一個擁有大約1800臺機器的集群上執行。每臺機器擁有兩個支持超線程的2GHz的Intel Xeon處理器，4GB內存，兩個160GB的IDE磁盤，和千兆以太網接入。這些機器被安排在一個二級樹形的交換網絡中，該網絡根部大約有100~200Gbps的聚合帶寬。所有機器都在同一個托管設施中，因此任何一對機器間的往返通信時間不超過1毫秒。 

Figure 2: Data transfer rate over time

雖然有4GB內存，但是大約1~1.5GB保留給了運行在集群上的其他任務。這些程序在一個周末的下午執行，此時CPUs，磁盤和網絡帶寬基本都空閑。

5.2 Grep

grep程序掃描了10^10個100字節的記錄，搜索一個相對稀有的三字符模式串(該模式串大約出現在92337個記錄中)。輸入被劃分為了大約64MB大小的片(M=15000)，整個輸出都放在了一個文件中(R=1)。

圖2展示了計算隨時間推移的進展。Y軸顯示了輸入數據的掃描速率。隨著安排到MapReduce計算的機器越來越多，速率也在逐步提升，當安排了1764個workers時速度達到峰值30GB/s以上。map任務結束后，速率來時下降且在大約80秒時到達0。整個計算從開始到結束大約花費了150秒。這包括1分鐘的啟動消耗。這個消耗來自向所有workers機器傳播程序、延遲與GFS的交互以開啟一組1000個輸入文件，和獲取局部優化所需的信息。