跳至主要內容
論文精煉 · Big data processing

MapReduce: Simplified Data Processing on Large Clusters

一個把平行化、容錯、資料局部性與負載平衡全部藏進函式庫的模型,使用者只需寫 map 與 reduce 兩個函式。

作者Jeffrey Dean、Sanjay Ghemawat,Google, Inc. 發表於OSDI 2004 年份2004
閱讀原始論文 PDF 所有論文

一口氣講完 — 整篇論文的濃縮

Google 的工程師寫過上百個一次性的分散式程式來處理爬取的網頁與請求日誌,而每一個程式都得重寫一遍資料切分、排程與故障處理。MapReduce 把這一切收斂成兩個使用者函式:map 把每筆輸入記錄轉成中間鍵值對,reduce 則把同一個鍵底下的所有值合併起來。一個 master 行程把輸入切成 M 份,把 map 與 reduce 任務指派給閒置的 worker,記錄每個 map 任務把中間檔案留在哪台機器的本機磁碟上,並在某個 worker 不再回應 ping 時重新執行它的任務。正因為模型受到限制、運算子通常具決定性,光靠重新執行就足以當作復原機制;同樣的限制也讓 master 能把任務排到資料所在的機器上,並對落後的任務額外啟動備份執行。在約 1800 台機器上,這套系統用約 150 秒 grep 完約一 TB 的資料,用 891 秒排序完一 TB,勝過當時公布最佳的 TeraSort 成績 1057 秒。

在這篇論文之前 — 它所降落的世界

到了 2003 年,Google 內部已有上百個特製運算,處理爬取的網頁、網站請求日誌與網頁連結圖,產出反向索引、各主機的頁面數統計,以及當日最熱門查詢的摘要。這些運算在概念上都很單純,但輸入資料大到必須分散到數百甚至數千台機器上才能在合理時間內跑完,於是每支程式都自己手寫一套切分、工作分配與故障復原的程式碼,把原本簡單的邏輯整個淹沒掉。硬體條件只讓事情更難:雙處理器的商用 x86 Linux 機器,2 至 4 GB 記憶體,掛的是便宜的 IDE 磁碟,單機網路是 100 Mb/s 到 1 Gb/s,但整體對分頻寬(bisection bandwidth)遠低於此,而叢集規模大到機器故障是家常便飯而非例外。當時可用的平行抽象,如 MPI、Bulk Synchronous Programming 與 parallel prefix 類模型,確實提高了表達層次,但多半只在小得多的規模上實作過,而且把機器故障丟回給程式設計師處理。前一年發表的 GFS 剛好讓人可以放心假設有一套複本化的檔案系統鋪在同一批本機磁碟上,這正是 MapReduce 賴以站立的底層。

問題 — 當時真正壞掉的地方

  • 這些運算的原始輸入大到必須把工作分散到數百甚至數千台機器上,才可能在合理時間內完成。
  • 為了平行化運算、分配資料與處理故障所寫的程式碼,會用大量複雜邏輯把原本簡單的運算徹底掩蓋掉。
  • 數百到數千台商用平價機器組成的叢集讓機器故障變成常態,因此一個要跑上好幾小時的函式庫必須能優雅地承受 worker 消失,而不是直接讓工作失敗。
  • 在這個環境裡網路頻寬是相對稀缺的資源:單機連線是 100 Mb/s 到 1 Gb/s,但整體對分頻寬平均低得多,天真地把中間資料搬來搬去根本擴展不了。
  • 單一台 straggler 機器就足以主宰整個工作的完成時間,原因可能是磁碟壞掉讓讀取效能從 30 MB/s 掉到 1 MB/s、機器上還跑著其他任務造成資源競爭,或是像作者遇過的機器初始化程式錯誤關掉了處理器快取,讓受影響機器慢上一百倍以上。
  • 完全沒有平行或分散式系統經驗的程式設計師根本無從利用叢集資源;而使用者程式在某些特定記錄上必然崩潰的臭蟲,也會讓整個運算永遠跑不完。

核心概念 — 主要貢獻,以及它們為何成立

刻意受限的程式模型

使用者只提供兩個函式,型別分別是 map (k1,v1) 到 list(k2,v2) 與 reduce (k2,list(v2)) 到 list(v2),其餘一切由函式庫掌管。這個限制本身就是重點:既然每筆記錄都獨立處理、每個鍵的值也獨立合併,執行環境就能自由決定要用幾台機器、每一塊在哪裡跑,以及什麼時候把同一塊跑兩次。同一個鍵的值是透過 iterator 而非串列交給 reduce 函式,因此單一鍵的值可以多到放不進記憶體。這可以看成是對既有受限平行模型的簡化與提煉,差別在於它附帶一個能擴展到數千個處理器的容錯實作。

以重新執行作為主要容錯手段

系統不對使用者狀態做檢查點,也不做訊息日誌,復原方式就是把任務再跑一次。master 週期性 ping 每個 worker,把沒有回應的標記為失效,並把它的任務全部退回 idle 狀態等待重新排程。已完成的 map 任務必須重跑,因為輸出留在死掉那台機器的本機磁碟上;已完成的 reduce 任務則不必重跑,因為輸出早已寫進全域檔案系統。這一切之所以成立,是因為模型讓任務不帶副作用且可重複執行,而這正是受限介面所保證的性質。

把運算搬到資料旁邊

輸入檔案以 64 MB 區塊存放在 GFS 中,通常有三份複本分佈在不同機器上,master 在指派 map 任務時會把這些位置資訊納入考量。它會盡量把 map 任務排到已經持有該輸入分片複本的機器上,若做不到,則退而求其次排到與複本同一台網路交換器下的機器。當一個大型 MapReduce 佔用叢集中相當大比例的 worker 時,多數輸入資料因此都是從本機磁碟讀取,完全不消耗網路頻寬。這個想法借鏡自 active disks 的研究,只是套用在直接掛了幾顆磁碟的商用機器上,而不是磁碟控制器上的處理器。

用備份任務對付 straggler

當運算接近尾聲時,master 會為所有仍在進行中的任務另外排一份備份執行,只要主要執行或備份執行任一完成,該任務就算完成。這等於把工作的尾巴從等待最慢那台機器,改成一場賽跑,而且完全不需要診斷機器為什麼變慢。這個機制經過調校,通常只多耗用幾個百分點的運算資源。效果非常明顯:關掉備份任務後,一 TB 的排序工作要多花 44% 的時間。

細粒度任務帶來動態負載平衡

M 與 R 都刻意取得遠大於 worker 機器數,因此每個 worker 在一個工作的生命週期裡會跑很多個任務。跑得快的機器自然會領走更多任務,動態負載平衡就此免費取得;而當某台機器掛掉時,它先前完成的大量 map 任務可以攤到所有其他 worker 上重跑,而不是集中在一處。正式環境常見的設定是在 2,000 台 worker 上使用 M = 200,000 與 R = 5,000。M 通常挑到讓每個 map 任務約涵蓋 16 MB 到 64 MB 的輸入,這同時也讓前述的局部性最佳化最有效。

用 combiner 縮小 shuffle 量

當 reduce 函式滿足交換律與結合律時,使用者可以註冊一個 combiner,在 map 所在的機器上先把中間記錄做部分合併,再送上網路。字數統計是最典型的例子:由於詞頻服從 Zipf 分佈,單一 map 任務會吐出成千上萬筆形如 the,1 的記錄,若不合併就得整批送往同一個 reduce 任務。通常 combiner 與 reducer 用的是同一份程式碼,唯一差別在於 combiner 的輸出寫進要送給 reduce 任務的中間檔案,而 reduce 的輸出寫進最終輸出檔。部分合併能明顯加速某幾類 MapReduce 運算。

運作方式 — 具體的機制

切分輸入與指派任務

使用者程式裡的函式庫先把輸入檔案切成 M 份,每份通常 16 MB 到 64 MB,可透過選用參數調整,接著在叢集上啟動許多份相同程式的副本。其中一份副本是特殊的 master,其餘都是 worker。總共有 M 個 map 任務與 R 個 reduce 任務待指派,master 會不斷挑一個閒置的 worker 指派其中一個任務。R 以及作用在中間鍵空間上的分割函式(預設為 hash(key) mod R)都由使用者指定。

Map 階段與本機落盤

被指派 map 任務的 worker 讀取對應的輸入分片,從中解析出鍵值對,逐一交給使用者的 Map 函式,並把產生的中間鍵值對先緩衝在記憶體中。系統會週期性地把緩衝內容寫到本機磁碟,並依分割函式切成 R 個區段,因此每個 map 任務最後會留下 R 塊檔案區段,一塊對應一個 reduce 任務。worker 再把這些區段在磁碟上的位置回報給 master。中間資料完全不寫進複本化的檔案系統,這正是省下網路頻寬的關鍵。

Shuffle、排序與 Reduce 階段

master 把中間檔案位置轉發給 reduce worker,後者以遠端程序呼叫(RPC)從各個 map worker 的本機磁碟把屬於自己的區段拉回來。當某個 reduce worker 讀完全部中間資料後,會依中間鍵排序,讓相同鍵相鄰;資料放不進記憶體時改用外部排序。之所以需要排序,是因為通常有很多不同的鍵會落到同一個 reduce 任務。接著 worker 掃過排好序的資料,對每個唯一鍵呼叫一次使用者的 Reduce 函式,傳入該鍵與其值的 iterator,並把結果附加到這個分割的最終輸出檔。由於每個分割是依鍵遞增順序處理,輸出檔天生有序,這也正是分散式排序得以成立、輸出可支援依鍵隨機查找的原因。

Master 的資料結構與記帳

master 為每個 map 與 reduce 任務保存狀態(idle、in-progress 或 completed),並為非 idle 的任務記下執行它的 worker 身分。對每個已完成的 map 任務,master 還保存該任務產生的 R 塊中間檔案區段的位置與大小,並在 reduce 任務進行中時把更新增量推送給對應的 worker;master 就是中間檔案位置從 map 端流向 reduce 端的唯一通道。這需要 O(M + R) 次排程決策與 O(M * R) 的記憶體狀態,不過常數很小,大約每一組 map 任務/reduce 任務對只佔一個位元組。所有任務完成後,master 喚醒使用者程式,MapReduce 呼叫就返回。

故障偵測與復原

master 週期性 ping 每個 worker,若一段時間內沒有回應就標記為失效;該 worker 上已完成的 map 任務,以及進行中的 map 或 reduce 任務,都會被重設為 idle 並重新排程。當某個 map 任務先由 worker A 執行、後來又由 worker B 重跑時,所有執行 reduce 任務的 worker 都會收到通知,尚未從 A 讀取資料的 reduce 任務就改向 B 讀。這套機制連相關性故障都撐得住:曾有一次正式運行中,網路維護讓每次約 80 台機器連續數分鐘無法連上,master 只是把這些機器的工作重新執行,運算持續向前推進並最終完成。master 本身則是例外:為它的資料結構做檢查點並不難,但既然只有一個 master、故障機率不高,目前的實作乾脆在 master 掛掉時中止整個運算,讓客戶端自行重試。

原子性提交與故障語意

每個進行中的任務都寫到自己的暫存檔:map 任務產生 R 個,reduce 任務產生一個。map 任務完成時,worker 把這 R 個暫存檔的名稱送給 master;若該 map 任務已被記錄為完成,master 直接忽略這則訊息,因此重複執行不會弄壞位置表。reduce worker 完成時,會把暫存輸出檔原子性地 rename 成最終輸出檔,倚賴檔案系統提供的原子 rename,保證即使同一個 reduce 任務在多台機器上跑過,最終檔案系統狀態也只含其中一次執行的資料。當運算子具決定性時,分散式執行的輸出與一次無故障的循序執行完全相同;當運算子不具決定性時,保證弱化為單一 reduce 任務層級的等價性,因為 R1 與 R2 各自提交的那次執行,可能讀到同一個 map 任務 M 的不同次執行結果。

運維層面的精修

系統提供一個選用模式,用來跳過會讓使用者程式必然崩潰的記錄:每個 worker 安裝 segmentation violation 與 bus error 的訊號處理器,在呼叫使用者函式前把目前引數的序號寫進全域變數,一旦收到訊號就送出一個帶著該序號的 last gasp UDP 封包給 master;當 master 看到同一筆記錄失敗超過一次,就在下次重新執行時指示跳過它。為了除錯,另有一份把整個工作在單機上循序跑完的函式庫實作,可以直接搭配 gdb 與剖析工具使用。master 內建一個 HTTP 伺服器,輸出狀態頁,顯示任務進度、輸入/中間/輸出的位元組數、處理速率、各任務標準錯誤與標準輸出的連結,以及哪些 worker 在跑哪些任務時失效。計數器機制讓使用者程式統計任意事件,數值搭著 ping 回應捎回 master,而 master 會扣除備份執行與重新執行造成的重複計數。

論文證明了什麼 — 量測數據與證明

  • 所有測試跑在約 1800 台機器上,每台配備兩顆 2 GHz Intel Xeon(開啟 HyperThreading)、4 GB 記憶體(其中 1 至 1.5 GB 被叢集上其他任務佔用)、兩顆 160 GB IDE 磁碟與一條 gigabit Ethernet;機器接在兩層樹狀交換網路下,根部可用的匯總頻寬約 100 至 200 Gbps,任兩台機器間的來回時間小於一毫秒。
  • 分散式 grep 掃過 10^10 筆 100 位元組的記錄,尋找一個只出現在 92,337 筆記錄中的三字元樣式,設定 M = 15000、R = 1;當 1764 個 worker 都投入後,輸入掃描速率尖峰超過 30 GB/s,整個運算約 150 秒完成,其中約有一分鐘是啟動開銷,來自把程式傳送到所有機器,以及與 GFS 互動開啟 1000 個輸入檔並取得局部性最佳化所需資訊。
  • 排序 10^10 筆 100 位元組記錄(約 1 TB)連同啟動開銷共花 891 秒,優於當時公布的 TeraSort 最佳成績 1057 秒;使用者程式不到 50 行,設定 M = 15000、R = 4000;輸入速率尖峰約 13 GB/s,shuffle 在約 600 秒完成,寫出在約 850 秒完成,由於輸出在 GFS 上做兩份複本,實際寫出 2 TB。
  • 在同一支排序程式上關掉備份任務後,960 秒時除了 5 個 reduce 任務外全部完成,但這幾個 straggler 又拖了 300 秒,整個工作拉長到 1283 秒,經過時間增加 44%。
  • 在排序開始數分鐘後,刻意殺掉 1746 個 worker 行程中的 200 個,圖上可見輸入速率轉為負值(已完成的 map 工作必須重做),但整個工作仍在 933 秒內完成,只比正常執行多 5%。
  • 2004 年 8 月,Google 共跑了 29,423 個 MapReduce 工作,平均完成時間 634 秒,耗用 79,186 機器日,讀入 3,288 TB 輸入、產生 758 TB 中間資料、寫出 193 TB 輸出,平均每個工作用 157 台 worker 機器並遇到 1.2 次 worker 死亡;用 MapReduce 重寫正式索引系統後,其中一個階段的程式碼從約 3800 行 C++ 縮到約 700 行。

限制與取捨 — 論文自承的,以及後來被發現的

  • 論文自承:整個系統只有一個 master,且沒有接手機制。為 master 的資料結構做檢查點並不困難,但既然單機故障機率不高,實作乾脆在 master 掛掉時中止整個運算,要求客戶端自行重試。
  • 論文自承:M 與 R 不能無限制放大,因為 master 必須做 O(M + R) 次排程決策,並在記憶體中保存 O(M * R) 的狀態;R 還會被使用者進一步壓低,因為每個 reduce 任務都會產生一個獨立的輸出檔。
  • 論文自承:強語意只在運算子具決定性時成立。不具決定性的 map 或 reduce 函式只能得到較弱的保證,即每個 reduce 任務各自對應某一次循序執行;而且系統不支援單一任務多個輸出檔的原子兩階段提交,因此有跨檔一致性需求的任務必須是決定性的。
  • 論文自承:把中間資料落盤要付出實際的 I/O 與頻寬代價。排序工作的 map 任務約有一半的時間與 I/O 頻寬花在把中間輸出寫到本機磁碟,而輸出階段還要寫兩份複本;論文也指出若底層檔案系統改用 erasure coding 而非複本,寫入所需的網路頻寬可以降低。
  • 後續研究揭露:map 接 shuffle 再接 reduce 的僵硬結構、加上每個階段邊界都落盤,對於疊代式與互動式工作負載太慢,而兩個函式的介面對多階段流水線來說又太底層。Dryad 把結構推廣成任意的資料流 DAG,Spark 以帶血統(lineage)追蹤的記憶體內 RDD 取代逐段落盤,Pig、Hive、FlumeJava 與 DryadLINQ 則在其上疊出更高階的語言。

它後來變成什麼 — 繼承這個想法的系統

MapReduce 成為之後十年批次處理的範本。Hadoop MapReduce 幾乎是逐項照抄地在 HDFS(GFS 的複製品)上重做了一遍,連對付 straggler 的推測執行與考量資料局部性的任務放置都一併繼承,並把這個模型變成開源大數據的預設平台。在它之上,Hive 與 Pig 把類 SQL 與資料流語言編譯成一連串 MapReduce 工作,而 FlumeJava、Cascading 與 DryadLINQ 則用流水線 API 把串接工作藏起來,等於承認原始的 map 與 reduce 對多階段任務而言太過底層。Dryad 把固定的兩階段結構推廣為任意資料流 DAG;Spark 保留了以決定性重新執行為核心的復原模型,但改成在記憶體內的 RDD 上追蹤血統,而不是每個階段都落盤,這才讓疊代式機器學習與互動式查詢在同一批叢集上變得實用。Google 自己也早已超越 MapReduce 的原始用途,先用 Percolator 的增量更新取代批次重建索引,之後再以 FlumeJava、Dataflow 與 Apache Beam 把批次與串流收攏在一起。真正被繼承下來的並不是那兩個函式的 API,而是一整套運維準則:考量局部性的排程、以細粒度任務達成動態負載平衡、用推測性備份執行處理尾端延遲,以及把決定性重新執行當成通用的復原機制,這些至今仍清楚地留在 YARN、Tez、Spark 與 Flink 之中。

論文原文 — 逐字引用

“使用者指定一個 map 函式,處理一組鍵值對並產生一批中間鍵值對;以及一個 reduce 函式,把所有與同一個中間鍵相關聯的中間值合併起來。”

Abstract

“我們採用具有使用者自訂 map 與 reduce 運算的函數式模型,使我們能輕易地平行化大型運算,並以重新執行作為容錯的主要機制。”

§1

“當一個 MapReduce 運算接近完成時,master 會為剩下仍在進行中的任務排定備份執行。只要主要執行或備份執行其中之一完成,該任務即標記為完成。”

§3.6

術語 — 依本篇論文的用法

Map
使用者撰寫的函式,接受一組輸入鍵值對,產生一批中間鍵值對,型別為 map (k1,v1) 到 list(k2,v2)。其輸出先緩衝在記憶體中,再落到該 worker 的本機磁碟。
Reduce
使用者撰寫的函式,接受一個中間鍵與該鍵所有值的 iterator,把它們合併成一組可能更小的值。通常每次呼叫只產生零個或一個輸出值。
Master
唯一一份不是 worker 的程式副本:負責把 map 與 reduce 任務指派給閒置的 worker、追蹤任務狀態,並把中間檔案區段的位置從 map 端轉送到 reduce 端。它同時也是故障偵測者與整個系統的單點故障。
Worker
使用者程式的一般副本,執行 master 指派給它的 map 或 reduce 任務。master 會週期性地 ping 它們,一旦停止回應就把其任務重新指派出去。
M 與 R
輸入分片數(即 map 任務數)與輸出分割數(即 reduce 任務數)。兩者都取得遠大於機器數,M 通常讓每個 map 任務涵蓋 16 至 64 MB,R 則取預期 worker 數的一個小倍數。
分割函式(partitioning function)
作用在中間鍵上、決定一筆記錄屬於 R 個 reduce 任務中哪一個的函式,預設為 hash(key) mod R。使用者可以自訂,例如只對 URL 鍵中的主機名稱做雜湊,讓同一主機的所有項目落在同一個輸出檔。
Combiner
選用的函式,在 map 所在機器上先把相同鍵的中間記錄做部分合併,再送上網路。適用於 reduce 函式滿足交換律與結合律的情況,通常與 reducer 共用同一份程式碼。
Straggler
在最後幾個 map 或 reduce 任務上花費異常久的機器,原因可能是磁碟即將損壞、機器上被排了其他任務造成資源競爭,或硬體與設定的錯誤。straggler 是拉長工作完成時間的主要元凶之一。
備份任務(backup task)
當運算接近完成時,master 為仍在進行中的任務額外排一份的重複執行。主要執行與備份執行誰先完成,該任務就算完成,代價經調校後只佔幾個百分點的額外資源。

在時間軸上的位置 — 這篇論文在整段故事中的座標

在時間軸上查看