MapReduce

論文《MapReduce: Simplified Data Processing on Large Clusters》筆記

Programming Model

概述

Map-Reduce是一種分布式系統的處理流程

按照論文圖中所示，Map階段實現的是對原始數據（raw data）的初步處理，可利于系統分發任務給reduce階段。而在Reduce階段，則從Intermediate files中讀取數據（可理解為從任務列表中索取任務）并產生輸出。

下面標注論文當中對Map、Reduce的解釋

Map, written by the user, takes an input pair and produces a set of intermediate key/value pairs. The MapReduce library groups together all intermediate values associated with the same intermediate key I and passes them to the Reduce function

The Reduce function, also written by the user, accepts
an intermediate key I and a set of values for that key. It
merges together these values to form a possibly smaller
set of values. Typically just zero or one output value is
produced per Reduce invocation. The intermediate values are supplied to the user’s reduce function via an iterator. This allows us to handle lists of values that are too
large to fit in memory.

從概念上講，Map主要負責從原始數據“提煉”出多個key-value對。同時MapReduce庫還要負責針對key對這些鍵值對進行分組（產生(key,array value[])這樣的數據，畢竟一個key可能對應復數個value），并傳遞給Reduce函數（傳遞key和value數組）

Reduce函數接受了來自Map的參數后，會將這些值進行合并，產生一個更小的數據集合。通常來說每次Reduce調用只產生0或者一個輸出值。論文中強調了Reduce接受array value[]時接受的是迭代器，以獲取超大占用value列表（廢話）

示例

給出的一個經典例子便是對大文檔內的單詞進行計數。

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

map函數中對文檔內容的單詞進行遍歷，并發送（word，“1”）的中間鍵值對。我們知道Map-Reduce是并發執行的，并不是Map執行一次，Reduce執行一次，這樣就成串行了

主要解釋為什么Reduce會接受一個value列表參數，畢竟執行Reduce的時候，指定word已經不止被遍歷到一遍了，那么它就會對應多個“1”。value參數列表由此而來。

Reduce接受到參數之后，就對value[]中的所有值進行求和，并發送出結果

在這里全部都是“1”,那result += ParseInt(v);就相當于result++;了

額外示例

URL訪問頻率計數

Map函數可以通過處理web頁面的請求日志來輸出鍵值對<URL,1>，Reduce接受到后處理（URL，array value[]）進行求和，發送<URL，total count>

Reverse Web-Link Graph（反向Web連接圖）

通常來講一個頁面會引用其他頁面（比如跳轉鏈接），則該頁面可成為source page，被引用的頁面被稱為target page。簡歷source page到target page的索引是很自然的事情，但同時也想記錄target page到source的關系圖。
因此Map函數可以為每一個在source page發現的target page輸出<target，source>鍵值對。Reduce將所有與target URL關聯的source連接起來，最終輸出<target,list(source)>

Inverted Index（倒排索引）

Inverted Index主要是建立關鍵詞對文檔的索引以應用在搜索引擎中。
Map函數解析每一個文檔，并且發送格式為<word, doucument ID>的序列。
Reduce函數接收到了指定單詞所有鍵值對后，開始連接所有的doucumet ID，并發射形為<word, list(document ID)>鍵值對

Implementation

Execution Overview（執行概述）

首先數據會被劃分，Map調用將被分發到多個機器上。輸入數據的分割可以由不同的機器并行處理。通過使用分區函數（例如hash(key) mod R）來將中間鍵值對進行分塊，然后被分發到多個Reduce調用上。分塊數和分區函數都可以由用戶指定。
用戶程序調用MapReduce函數時，會發生以下一系列動作：

• 在用戶程序中的MapReduce庫會把輸入文件分成 M 塊，通常為16-64M不等（可以由用戶指定）。然后在 一組 機器上啟動該程序的許多副本(RPC調用)

• Master單元（母單元）負責分配任務到工作單元，會選擇空閑的工作單元去處理Map或Reduce任務。

• 被分配到map任務的工作單元讀取相應的輸入分塊內容。它將解析數值對，并將每一個鍵值對傳遞給用戶定義的Map函數(map（key, value）)。Map函數產生的中間鍵值對存放在內存緩沖區。

• 周期性地將緩沖區內的鍵值對寫入本地磁盤，并通過分區函數劃分為R個區域。這些鍵值對在本地磁盤上的位置被傳遞給Master單元，Master負責將這些位置轉發給reduce worker。

• 當Master單元體制Reduce單元這些位置時，將使用RPC調用從map worker的本地磁盤讀取緩沖數據。當Reduce單元讀取了所有的中間數據后，它將所有的中間鍵值對進行排序，以便將所有重復出現的相同鍵值對分組在一起。排序是必要的，因為通常有許多不同的鍵映射到相同的reduce任務。果中間數據量太大，內存無法容納，則使用外部排序。

• Reduce單元遍歷已排序的中間數據，并且對于每一個唯一的“key”都會將key和相應的 intermediate values[]傳遞給用戶定義的Reduce調用(Reduce_function(key, value[]))。Reduce函數的輸出將被附加到這個Reduce分區的最終輸出文件中。

• 當所有的map和reduce任務完成后，master單元會喚起user program.
  此時，用戶程序中的MapReduce調用返回到用戶代碼。

Master Data Structures

Master單元擁有幾個數據結構，用于存儲每個map和reduce任務的狀態 (idle, in-progress,or completed)和工作機器的標識(for non-idle tasks)
Matser單元是將本地中間文件分塊從map task傳遞到reduce task的通道。因此對于每一個完成了的map task，Master單元將儲存 R intermediate file regions 的位置和 sizes。當Map任務完成時，會收到對該位置和大小信息的更新。這些信息會被遞增地推送到正在執行reduce任務的工作單元。

Fault Tolerance（容錯性）

主節點定期向每個工作節點發送ping請求。如果在規定的時間內沒有收到某個工作節點的響應，主節點會標記該節點為失敗。該工作節點已經完成的任何map任務會被重置回初始的空閑狀態，從而可以在其他工作節點上重新調度。同樣，任何在失敗的工作節點上進行中的map任務或reduce任務也會被重置為空閑狀態，并可重新調度

已完成的map任務會在失敗后重新執行，因為它們的輸出存儲在失敗機器的本地磁盤上，因此無法訪問。而已完成的reduce任務無需重新執行，因為它們的輸出存儲在全局文件系統中。

當一個map任務首先由工作節點A執行，然后因為A失敗，任務被工作節點B重新執行時，所有執行reduce任務的工作節點都會收到重執行的通知。任何尚未從工作節點A讀取數據的reduce任務將會從工作節點B讀取數據。

MapReduce對大規模的工作節點故障具有很強的恢復能力。例如，在一次MapReduce操作中，網絡維護導致運行中的集群中每次有80臺機器變得不可訪問，持續幾分鐘。MapReduce的主節點簡單地重新執行了這些無法訪問的工作節點的任務，繼續向前推進，最終完成了MapReduce操作。

Master Failure

使主節點定期寫入上述主節點數據結構的檢查點是比較簡單的。如果主節點任務崩潰，可以從最后的檢查點狀態重新啟動新的副本。然而，考慮到只有一個主節點，其故障的可能性較小，因此我們當前的實現會在主節點失敗時中止MapReduce計算。客戶端可以檢測到這種情況，并在需要時重試MapReduce操作

Semantics in the Presence of Failures

當用戶提供的 map 和 reduce 操作是輸入值的確定性（deterministic）函數時，我們的分布式實現能夠產生與無故障順序執行整個程序時相同的輸出。

• 確定性（deterministic）函數：相同輸入能得到相同的輸出
• 非確定性（non-deterministic）函數：相同輸入可能會得到不同輸出（比如擁有隨機性的函數）

我們依賴 map 和 reduce 任務輸出的原子提交（atomic commit）來保證這一特性。每個正在執行的任務會將其輸出寫入私有的臨時文件：

• 一個 reduce 任務生成一個臨時文件。
• 一個 map 任務生成 R 個臨時文件（每個 reduce 任務對應一個文件）。

當 map 任務完成時，工作節點會向主節點發送一條消息，并在其中包含 R 個臨時文件的名稱。如果主節點收到的 map 任務完成消息對應于一個已經完成的任務，則會忽略該消息。否則，它會將 R 個文件名記錄到主節點的數據結構中。

當 reduce 任務完成時，其工作節點會原子地將臨時輸出文件重命名為最終輸出文件。如果同一個 reduce 任務在多個機器上執行，則可能會有多個重命名操作針對同一個最終輸出文件。我們依賴底層文件系統提供的原子重命名操作（atomic rename）來確保最終的文件系統狀態只包含 reduce 任務的其中一次執行所產生的數據。

絕大多數 map 和 reduce 操作都是確定性的，在這種情況下，我們的語義等價于順序執行，使程序員更容易推理程序的行為。然而，當 map 或 reduce 操作是非確定性（non-deterministic）時，我們提供的是較弱但仍合理的語義。

在存在非確定性操作的情況下，一個特定的 reduce 任務 R1 的輸出等價于該任務在某次順序執行中產生的結果。然而，另一個 reduce 任務 R2 的輸出可能對應于不同的順序執行中產生的結果。

考慮 map 任務 M 和 reduce 任務 R1、R2，定義 e(Ri) 為 Ri 的最終提交執行（每個 reduce 任務最終只有一次成功的提交執行）。較弱的語義產生的原因如下：

• e(R1) 可能讀取的是 M 在某次執行中產生的輸出，而
• e(R2) 可能讀取的是 M 在不同執行中產生的輸出。

因此，在 map 任務 M 被重新執行的情況下，不同的 reduce 任務可能讀取 M 在不同執行中產生的輸出，從而導致 reduce 任務的最終結果可能來自不同的 map 執行版本。

Locality（數據局部性）

在我們的計算環境中，網絡帶寬是一種相對稀缺的資源。為了節省網絡帶寬，我們充分利用輸入數據存儲在集群機器的本地磁盤這一特性（這些數據由 GFS [Google File System] 管理）。

GFS 將每個文件劃分為 64MB 的塊，并在不同的機器上存儲多個副本（通常是 3 份副本）。MapReduce 主節點會根據輸入文件的位置信息，優先將 map 任務調度到包含該任務所需數據副本的機器上。

如果無法將任務調度到存儲有該數據副本的機器上，主節點會盡量選擇靠近數據副本的機器（例如，調度到同一網絡交換機下的其他工作節點）。

當在一個大規模的集群上運行 大規模 MapReduce 任務時，大部分輸入數據都可以在本地讀取，不會占用網絡帶寬，從而提高任務執行效率。

Task Granularity（任務粒度）

在 MapReduce 計算框架中，我們將 map 階段 劃分為 M 個任務，將 reduce 階段 劃分為 R 個任務。

1. M 和 R 的選擇

• 理想情況下，M 和 R 的數量應遠大于工作節點的數量（worker machines）。

• 這樣做的好處：
  
  1.提高動態負載均衡能力——任務可以更均勻地分配給不同的 worker 機器，避免部分機器過載或空閑。
  
  2.加速故障恢復——如果某個 worker 失效，它已完成的多個 map 任務可以分散到其他 worker 機器上執行，減少整體恢復時間。

2. M 和 R 過大的限制
   調度開銷：主節點（master）需要進行 O(M + R) 次任務調度決策。
   內存占用：主節點存儲 O(M × R) 的任務狀態信息（不過，實際內存占用較小，每個 map-reduce 任務對大約存儲 1 字節 的狀態信息）。
   R 的用戶限制：
   每個 reduce 任務最終會生成一個單獨的輸出文件，所以 R 不能太大，否則會產生過多的小文件，影響后續處理。
3. 任務粒度的優化
   通常，我們選擇 M 使得每個任務處理的輸入數據大小在 16MB 到 64MB 之間。
   這樣可以保證數據局部性（Locality Optimization）最大化，減少遠程數據傳輸。
   R 一般設置為 worker 機器數量的一個小倍數，以保證 reduce 任務能夠較好地并行執行。
   例如，在 2,000 臺 worker 機器 的計算環境中，我們常用的參數是：
   M = 200,000
   R = 5,000
   這種配置能充分利用集群計算資源，同時保持合理的調度開銷，提高任務執行效率。