MapReduce論文閱讀記錄

本文為閱讀MapReduce論文的記錄，内容主要是論文的第三部分——實作。友善本人今後檢視。

1. 運作概述

下圖展示了 MapReduce 過程的整體情況

當使用者程式執行 MapReduce 時，會依次發生以下動作（對應圖中的标号）：

1. 使用者程式中的 MapReduce 庫将輸入檔案分成 M 個分片，每片有16M-64M（由使用者決定），MapReduce 庫還會将程式拷貝到叢集機器上。
2. 叢集中有一個 master，多個 worker。在拷貝程式過程中，其中 master 獲得的程式是特殊的。master 将配置設定工作給 worker。現在有 M 個 map 任務和 R 個 reduce 任務需要被配置設定。master 會選擇空閑的 worker，配置設定給 map 任務或 reduce 任務。一個 worker 隻能擔任一個 map 任務或 一個reduce 任務。
3. 被配置設定 map 任務的 worker 接下來會讀取相應的輸入塊，将輸入資料解析成 k-v 對，并将 k-v 對傳給使用者定義的 Map 函數。由 Map 函數生成的中間結果 k-v 對緩存在記憶體中。
4. 緩存的中間結果将定期地被寫到本地磁盤上。分區函數（例如，hash(key) mod R）将中間結果分割成 R 個分區。然後，中間結果在本地磁盤的位置将傳回給 master，接着 master 将負責把這些位置傳給reduce worker。

5. 當 reduce worker 被 master 通知了中間結果的位置，它将通過 RPC 讀取 map worker 本地磁盤上的中間結果。當完成讀取工作，它會對中間結果進行排序，讓具有相同 key 的對被分組在一塊。

   排序工作的重要性在于：通常具有不同 key 的對會被分到同一個 reduce 任務中（與分區函數有關）。如果由于中間結果過大，無法裝進記憶體進行排序，需要使用外部排序。

6. reduce worker 對已排序的資料進行周遊，每遇到一個不同的 key，便将 key 與對應的一系列 value 傳給使用者定義的 reduce 函數。其輸出将作為該 reduce 分區的結果，追加到最終的輸出檔案中。
7. 當所有的 map 、reduce 任務完成， master 将喚醒使用者程式。同時，使用者程式中的mapreduce 調用得到傳回。

在執行完成後，mapreduce 的輸出将是 R 個檔案（每個 reduce 任務一個）。通常，使用者不需要将這 R 個檔案合并成一個，可作為輸入傳給另一個 mapreduce 調用，或另一個分布式程式。

2. Master 資料結構

對于每個 map、reduce 任務，master 都會存儲其狀态（idle、in-progress、completed）和 non-idle的 worker 的資訊。

master 在 map 任務到 reduce 任務之間傳輸中間結果的位置。對于每個完成的 map 任務，master 會存儲其 R 個分區的位置和大小，并将該資訊逐漸傳輸給處于 in-progress的reduce worker。

3. 容錯

3.1 worker 故障

master 定期地 ping 所有 worker。如果一個 worker 長時間沒有響應， master 認為該 worker 已故障。該worker 上，以下任務，将被重置為 idle 狀态，并将該任務重新配置設定到其他 worker 上

• 處于 completed 狀态的 map 任務
• 處于 in-progress 狀态的 map、reduce 任務
• 處于 in-progress 狀态的 reduce 任務

completed 狀态的 map 任務需要重新執行的原因：輸出存儲在故障機器的本地磁盤上，已經不可通路了。

completed 狀态的 reduce 任務不需要重新執行的原因：輸出存儲在全局檔案系統（GFS）上。

worker A 執行 map 任務，由于 A 故障了，接着由 worker B 執行該 map 任務。所有在運作 reduce 任務的 worker 都将被通知重新執行，而還沒有從 worker A 讀資料的 reduce 任務，将轉為 worker B。

3.2 master 故障

master 定期檢查點記錄狀态，當 master 任務死亡時，從最近的檢查點狀态開始執行。

3.3 本地性

網絡帶寬是我們的計算環境中相對稀缺的資源。 我們通過利用輸入資料（由 GFS 管理）存儲在組成我們叢集的機器的本地磁盤上的來節省網絡帶寬。 GFS 将每個檔案分成 64MB 塊，并在不同的機器上存儲每個塊的多個副本（通常是3個副本）。 mapReduce master 将輸入檔案的位置資訊考慮在内，并嘗試在包含相應輸入資料副本的機器上配置設定 map 任務。否則，它将嘗試在該任務的輸入資料副本附近（例如，在與包含資料的計算機處于同一網絡交換機上的機器上）安排一個 map 任務。 在叢集中大部分 worker 上運作大型MapReduce操作時，大多數輸入資料都是本地讀取的，不會消耗網絡帶寬。

3.4 任務粒度

從上文我們可以得知，map 階段被劃分成 M 個 task，reduce 階段被劃分成 R 個 task，M 和 R 一般會比叢集中節點數大得多。每個節點運作多個 task 有利于動态的負載均衡，加速 worker 從失敗中恢複。

在具體的實作中，M 和 R 的大小是有實際限制的，因為 master 至少要做 O(M＋R) 次的排程決策，并且需要保持O(M * R)個狀态（使用的記憶體并不大，一條 M-R 記錄需要 1 位元組）。

通常情況下，R 的大小是由使用者指定的，而對 M 的選擇要保證每個任務的輸入資料大小，即一個輸入分片在 16MB～64MB 之間（資料本地性最優）。R 的大小是 worker 數量的一個較小的倍數。

3.5 備份任務

一種最常見的延長 mapreduce 運作總時間的原因是 “straggler”：一台機器花費異常時間完成最後一個 map 或 reduce 任務。“straggler” 出現的原因有很多，例如：磁盤有問題，讀取速度下降；叢集排程在機器上安排了其他任務，由于競争CPU、記憶體、本地磁盤或網絡帶寬，導緻其更慢地執行 mapreduce 代碼。

解決“straggler”的機制：當 mapreduce 操作快完成時， master 會備份剩餘的 in-progress 狀态的任務。無論主程式或備份程式執行完成，該任務都會被标記為已完成。