經典分佈式論文閱讀：MapReduce

本文是MapReduce論文的學習筆記。MapReduce框架須要使用者提供map和reduce函數，map函數將一個鍵/值輸入轉換成一組中間鍵/值，而後reduce函數將全部具備相同鍵的中間鍵值對進行合併，而其餘的任務調度、文件分割、容錯處理等工做都由框架來完成。

編程模型

MapReduce框架用戶須要提供兩個函數map和reduce：

• Map：將輸入的一對鍵值對轉換爲一組中間鍵值對。
• Reduce：將全部鍵相同的中間鍵值對合並，獲得關於那個鍵的結果。

簡單的例子

初看很難體會到MapReduce的設計初衷，而MapReduce最經典的例子就是單詞計數任務，單詞計數任務的兩個函數定義以下：

func map(filename string, contents string) []mapreduce.KeyValue {
	words := strings.FieldsFunc(contents, func(r rune) bool {
		return !unicode.IsLetter(r)
	})
	var res []mapreduce.KeyValue
	for _, word := range words {
		res = append(res, mapreduce.KeyValue{ Key:word, Value:"1" })
	}
	return res
}

func reduce(key string, values []string) string {
	result := 0
	for _, value := range values {
		count, _ := strconv.Atoi(value)
		result += count
	}
	return strconv.Itoa(result)
}
複製代碼

上述例子中map函數輸入鍵爲文件名，輸入值爲文件內容，map函數將文件內容分割爲多個單詞，中間鍵值對爲單詞和單詞出現次數「1」，而reduce函數將某個單詞全部出現的次數相加。

更多的例子

除了最簡單的單詞統計之外，還有不少的問題均可以套用MapReduce的模型解決。

• 分佈式grep：Map函數在某一行匹配成功以後產生一箇中間鍵值對，reduce函數將匹配結果簡單合併。
• URL訪問統計：Map函數根據每一條訪問日誌產生一箇中間鍵值對<URL,1>，reduce函數將URL的全部中間鍵值對的值相加，產生結果<URL,訪問次數>。
• 反向網頁連接圖：當來源網頁中出現一次目標連接，map函數產生一箇中間鍵值對<目標,來源>。Reduce函數合併相同目標的中間值，產生<目標,list(來源)>。
• 反向索引：Map函數解析文檔後，產生如<單詞,文檔編號>的中間鍵值對，而後reduce函數合併中間鍵值對，產生結果<單詞,list(文檔編號)>。最終結果組成一個反向索引，能夠用於查詢單詞出現的文檔。
• 分佈式排序：Map函數根據每一條記錄中參與排序的鍵取出，產生中間結果<鍵,記錄>，reduce函數則原樣輸出中間鍵值對便可。

系統實現

運行流程

MapReduce須要處理的數據會被事先分割爲M片斷，中間數據被分配給R個片斷，分割過程由分割函數hash(鍵) mod R，分片數M和R以及哈希函數都由用戶定義。

1. MapReduce將輸入文件分割爲M個片斷。
2. 計算集羣由一個master和多個worker組成，master負責將map或者reduce任務分配給worker完成。
3. 分配到map任務的worker讀取輸入文件片斷，從片斷中解析出鍵值對傳遞給用戶定義的map函數獲得一組中間鍵值對保存到內存中。
4. 內存中的中間鍵值對會被週期性地寫入本地磁盤，而後被分割爲R個片斷，將片斷的保存位置通知master用於reduce任務。
5. 分配到reduce任務的worker讀取來自map任務worker磁盤的中間鍵值對。當讀取徹底部的中間數據以後，將全部的鍵值對按照鍵順序排序，將鍵值相同的值合爲一組。
6. 執行reduce任務的worker將相同鍵的中間值集合傳遞給用戶定義的reduce函數，將輸出添加到當前reduce任務對應的片斷中。
7. 當所有的map和reduce操做完成以後，MapReduce通知用戶程序處理輸出文件。

MapReduce處理結果一般會被保存到R個文件片斷中，文件片斷一般不須要被合併，直接用於其餘的分佈式任務。

容錯處理

Worker錯誤

Master會不斷ping各個worker，若是某個worker產生錯誤，那麼會被重置到可調度狀態。發生錯誤的時候，已經完成的map任務須要被從新執行，由於map的結果保存到本地磁盤中，而已完成的reduce任務不須要被從新執行，由於reduce任務的結果被寫入全局文件系統，

Master錯誤

可讓master按期將狀態保存到磁盤，崩潰後直接利用保存的狀態恢復。另外，也能夠考慮直接在master奔潰的時候終止MapReduce任務。

系統優化

在實現MapReduce實現中，由不少的技巧能夠提升系統的運行效率。

• 局部性：若是MapReduce的輸入文件保存在分佈式文件系統（例如GFS）中，那麼能夠結合分佈式文件系統，將map任務分配給保存有輸入文件分片的worker，或者退而求其次選擇裏文件分片保存位置最近的worker。
• 細粒度：輸入文件分割數目M一般要比worker大不少，這樣方便系統進行更加合理地調度，原文中建議每一個分割大小爲16MB到64MB， 可是如今已是快15年後了。
• 備份任務：在執行MapReduce任務時，會有一些worker因爲各類緣由致使任務執行很是慢。所以，當MapReduce任務快要結束的時候，系統將那些還沒有完成的任務分配給其餘worker同時執行，來加快完成速度。
• 合併函數：在單詞統計任務中，咱們知道單詞是符合Zipf分佈的，所以 會有大量的例如<the,1>這樣經常使用詞產生的中間數據傳遞給某個reduce任務的worker，這會給某個worker帶來巨大的負擔。解決的方法就是讓map執行完成後調用合併函數處理一編中間數據。合併函數和reduce函數一般是同樣的，只是調用場景不一樣。

參考文獻

1. Dean, Jeffrey, and Sanjay Ghemawat. "MapReduce: simplified data processing on large clusters." Communications of the ACM 51.1 (2008): 107-113.