Google MapReduce中文版

摘要

1、介紹

2、編程模型

2.1、例子

2.2、類(lèi)型

（alex注：原文中這個(gè)domain的含義不是很清楚，我參考Hadoop、KFS等實(shí)現，map和reduce都使用了泛型，因此，我把domain翻譯成類(lèi)型推導的域）。
我們的C++中使用字符串類(lèi)型作為用戶(hù)自定義函數的輸入輸出，用戶(hù)在自己的代碼中對字符串進(jìn)行適當的類(lèi)型轉換。

2.3、更多的例子

3、實(shí)現

3.1、執行概括

在成功完成任務(wù)之后，MapReduce的輸出存放在R個(gè)輸出文件中（對應每個(gè)Reduce任務(wù)產(chǎn)生一個(gè)輸出文件，文件名由用戶(hù)指定）。一般情況下，用戶(hù)不需要將這R個(gè)輸出文件合并成一個(gè)文件–他們經(jīng)常把這些文件作為另外一個(gè)MapReduce的輸入，或者在另外一個(gè)可以處理多個(gè)分割文件的分布式應用中使用。

3.2、Master數據結構

3.3、容錯

當worker故障時(shí)，由于已經(jīng)完成的Map任務(wù)的輸出存儲在這臺機器上，Map任務(wù)的輸出已不可訪(fǎng)問(wèn)了，因此必須重新執行。而已經(jīng)完成的Reduce任務(wù)的輸出存儲在全局文件系統上，因此不需要再次執行。

使用MapReduce模型的程序員可以很容易的理解他們程序的行為，因為我們絕大多數的Map和Reduce操作是確定性的，而且存在這樣的一個(gè)事實(shí)：我們的失效處理機制等價(jià)于一個(gè)順序的執行的操作。當Map或/和Reduce操作是不確定性的時(shí)候，我們提供雖然較弱但是依然合理的處理機制。當使用非確定操作的時(shí)候，一個(gè)Reduce任務(wù)R1的輸出等價(jià)于一個(gè)非確定性程序順序執行產(chǎn)生時(shí)的輸出。但是，另一個(gè)Reduce任務(wù)R2的輸出也許符合一個(gè)不同的非確定順序程序執行產(chǎn)生的R2的輸出。

3.4、存儲位置

3.5、任務(wù)粒度

但是實(shí)際上，在我們的具體實(shí)現中對M和R的取值都有一定的客觀(guān)限制，因為master必須執行O(M+R)次調度，并且在內存中保存O(M*R)個(gè)狀態(tài)（對影響內存使用的因素還是比較小的：O(M*R)塊狀態(tài)，大概每對Map任務(wù)/Reduce任務(wù)1個(gè)字節就可以了）。

3.6、備用任務(wù)

4、技巧

雖然簡(jiǎn)單的Map和Reduce函數提供的基本功能已經(jīng)能夠滿(mǎn)足大部分的計算需要，我們還是發(fā)掘出了一些有價(jià)值的擴展功能。本節將描述這些擴展功能。

4.1、分區函數

4.2、順序保證

4.3、Combiner函數

部分的合并中間結果可以顯著(zhù)的提高一些MapReduce操作的速度。附錄A包含一個(gè)使用combiner函數的例子。

4.4、輸入和輸出的類(lèi)型

Reader并非一定要從文件中讀取數據，比如，我們可以很容易的實(shí)現一個(gè)從數據庫里讀記錄的Reader，或者從內存中的數據結構讀取數據的Reader。

類(lèi)似的，我們提供了一些預定義的輸出數據的類(lèi)型，通過(guò)這些預定義類(lèi)型能夠產(chǎn)生不同格式的數據。用戶(hù)采用類(lèi)似添加新的輸入數據類(lèi)型的方式增加新的輸出類(lèi)型。

4.5、副作用

如果一個(gè)任務(wù)產(chǎn)生了多個(gè)輸出文件，我們沒(méi)有提供類(lèi)似兩階段提交的原子操作支持這種情況。因此，對于會(huì )產(chǎn)生多個(gè)輸出文件、并且對于跨文件有一致性要求的任務(wù)，都必須是確定性的任務(wù)。但是在實(shí)際應用過(guò)程中，這個(gè)限制還沒(méi)有給我們帶來(lái)過(guò)麻煩。

4.6、跳過(guò)損壞的記錄

每個(gè)worker進(jìn)程都設置了信號處理函數捕獲內存段異常（segmentation violation）和總線(xiàn)錯誤（buserror）。在執行Map或者Reduce操作之前，MapReduce庫通過(guò)全局變量保存記錄序號。如果用戶(hù)程序觸發(fā)了一個(gè)系統信號，消息處理函數將用“最后一口氣”通過(guò)UDP包向master發(fā)送處理的最后一條記錄的序號。當master看到在處理某條特定記錄不止失敗一次時(shí)，master就標志著(zhù)條記錄需要被跳過(guò)，并且在下次重新執行相關(guān)的Map或者Reduce任務(wù)的時(shí)候跳過(guò)這條記錄。

4.7、本地執行

4.8、狀態(tài)信息

另外，處于最頂層的狀態(tài)頁(yè)面顯示了哪些worker失效了，以及他們失效的時(shí)候正在運行的Map和Reduce任務(wù)。這些信息對于調試用戶(hù)代碼中的bug很有幫助。

4.9、計數器

map(String name, String contents):
 for each word w in contents:
  if (IsCapitalized(w)):
   uppercase->Increment();
  EmitIntermediate(w, “1″);

計數器機制對于MapReduce操作的完整性檢查非常有用。比如，在某些MapReduce操作中，用戶(hù)需要確保輸出的key value pair精確的等于輸入的key value pair，或者處理的German文檔數量在處理的整個(gè)文檔數量中屬于合理范圍。

5、性能

5.1、集群配置

在4GB內存里，大概有1-1.5G用于運行在集群上的其他任務(wù)。測試程序在周末下午開(kāi)始執行，這時(shí)主機的CPU、磁盤(pán)和網(wǎng)絡(luò )基本上處于空閑狀態(tài)。

5.2、GREP

這個(gè)分布式的grep程序需要掃描大概10的10次方個(gè)由100個(gè)字節組成的記錄，查找出現概率較小的3個(gè)字符的模式（這個(gè)模式在92337個(gè)記錄中出現）。輸入數據被拆分成大約64M的Block（M=15000），整個(gè)輸出數據存放在一個(gè)文件中（R=1）。

圖2顯示了這個(gè)運算隨時(shí)間的處理過(guò)程。其中Y軸表示輸入數據的處理速度。處理速度隨著(zhù)參與MapReduce計算的機器數量的增加而增加，當1764臺worker參與計算的時(shí)，處理速度達到了30GB/s。當Map任務(wù)結束的時(shí)候，即在計算開(kāi)始后80秒，輸入的處理速度降到0。整個(gè)計算過(guò)程從開(kāi)始到結束一共花了大概150秒。這包括了大約一分鐘的初始啟動(dòng)階段。初始啟動(dòng)階段消耗的時(shí)間包括了是把這個(gè)程序傳送到各個(gè)worker機器上的時(shí)間、等待GFS文件系統打開(kāi)1000個(gè)輸入文件集合的時(shí)間、獲取相關(guān)的文件本地位置優(yōu)化信息的時(shí)間。

5.3、排序

在這個(gè)benchmark測試中，我們使用的分區函數知道key的分區情況。通常對于排序程序來(lái)說(shuō)，我們會(huì )增加一個(gè)預處理的MapReduce操作用于采樣key值的分布情況，通過(guò)采樣的數據來(lái)計算對最終排序處理的分區點(diǎn)。

還有一些值得注意的現象：輸入數據的讀取速度比排序速度和輸出數據寫(xiě)入磁盤(pán)速度要高不少，這是因為我們的輸入數據本地化優(yōu)化策略起了作用 —絕大部分數據都是從本地硬盤(pán)讀取的，從而節省了網(wǎng)絡(luò )帶寬。排序速度比輸出數據寫(xiě)入到磁盤(pán)的速度快，這是因為輸出數據寫(xiě)了兩份（我們使用了2路的GFS文件系統，寫(xiě)入復制節點(diǎn)的原因是為了保證數據可靠性和可用性）。我們把輸出數據寫(xiě)入到兩個(gè)復制節點(diǎn)的原因是因為這是底層文件系統的保證數據可靠性和可用性的實(shí)現機制。如果底層文件系統使用類(lèi)似容錯編碼[14](erasurecoding)的方式而不是復制的方式保證數據的可靠性和可用性，那么在輸出數據寫(xiě)入磁盤(pán)的時(shí)候，就可以降低網(wǎng)絡(luò )帶寬的使用。

5.4、高效的backup任務(wù)

5.5、失效的機器

6、經(jīng)驗

在每個(gè)任務(wù)結束的時(shí)候，MapReduce庫統計計算資源的使用狀況。在表1，我們列出了2004年8月份MapReduce運行的任務(wù)所占用的相關(guān)資源。

6.1、大規模索引

7、相關(guān)工作

TACC[7]是一個(gè)用于簡(jiǎn)化構造高可用性網(wǎng)絡(luò )服務(wù)的系統。和MapReduce一樣，它也依靠重新執行機制來(lái)實(shí)現的容錯處理。

8、結束語(yǔ)

我們也從MapReduce開(kāi)發(fā)過(guò)程中學(xué)到了不少東西。首先，約束編程模式使得并行和分布式計算非常容易，也易于構造容錯的計算環(huán)境；其次，網(wǎng)絡(luò )帶寬是稀有資源。大量的系統優(yōu)化是針對減少網(wǎng)絡(luò )傳輸量為目的的：本地優(yōu)化策略使大量的數據從本地磁盤(pán)讀取，中間文件寫(xiě)入本地磁盤(pán)、并且只寫(xiě)一份中間文件也節約了網(wǎng)絡(luò )帶寬；第三，多次執行相同的任務(wù)可以減少性能緩慢的機器帶來(lái)的負面影響（alex注：即硬件配置的不平衡），同時(shí)解決了由于機器失效導致的數據丟失問(wèn)題。

9、感謝

10、參考資料

附錄A、單詞頻率統計

// User’s map function
class WordCounter : public Mapper {
 public:
  virtual void Map(const MapInput& input) {
   const string& text = input.value();
   const int n = text.size();
   for (int i = 0; i < n; ) {
    // Skip past leading whitespace
    while ((i < n) && isspace(text[i]))
     i++;

   // Find word end
   int start = i;
   while ((i < n) && !isspace(text[i]))
    i++;
   if (start < i)
    Emit(text.substr(start,i-start),”1″);
  }
 }
};

REGISTER_MAPPER(WordCounter);

// User’s reduce function
class Adder : public Reducer {
 virtual void Reduce(ReduceInput* input) {
  // Iterate over all entries with the
  // same key and add the values
  int64 value = 0;
  while (!input->done()) {
   value += StringToInt(input->value());
   input->NextValue();
  }

  // Emit sum for input->key()
  Emit(IntToString(value));
 }
};

REGISTER_REDUCER(Adder);

int main(int argc, char** argv) {
 ParseCommandLineFlags(argc, argv);
 
 MapReduceSpecification spec;
 
 // Store list of input files into “spec”
 for (int i = 1; i < argc; i++) {
  MapReduceInput* input = spec.add_input();
  input->set_format(“text”);
  input->set_filepattern(argv[i]);
  input->set_mapper_class(“WordCounter”);
 }

 // Specify the output files:
 // /gfs/test/freq-00000-of-00100
 // /gfs/test/freq-00001-of-00100
 // …
 MapReduceOutput* out = spec.output();
 out->set_filebase(“/gfs/test/freq”);
 out->set_num_tasks(100);
 out->set_format(“text”);
 out->set_reducer_class(“Adder”);
 
 // Optional: do partial sums within map
 // tasks to save network bandwidth
 out->set_combiner_class(“Adder”);

 // Tuning parameters: use at most 2000
 // machines and 100 MB of memory per task
 spec.set_machines(2000);
 spec.set_map_megabytes(100);
 spec.set_reduce_megabytes(100);
 
 // Now run it
 MapReduceResult result;
 if (!MapReduce(spec, &result)) abort();
 
 // Done: ‘result’ structure contains info
 // about counters, time taken, number of
 // machines used, etc.
 return 0;
}