原文發表于 2006 年 11 月 15 日

　　微軟著名的 C++大師 Herb Sutter 在 2005 年初的時候曾經寫過一篇重量級的文章——The Free Lunch Is Over: A Fundamental Turn Toward Concurrency in Software，預言OO之后軟件開發將要面臨的又一次重大變革——并行計算。

　　摩爾定律統治下的軟件開發時代有一個非常有意思的現象：“Andy giveth, and Bill taketh away.”。不管CPU的主頻有多快，我們始終有辦法來利用它，而我們也陶醉在機器升級帶來的程序性能提高中。

　　我記著我大二的時候曾經做過一個五子棋的程序，當時的算法就是預先設計一些棋型（有優先級），然后掃描棋盤，對形勢進行分析，看看當前走哪部對自己最重要。當然下棋還要堵別人，這就需要互換雙方的棋型再計算。如果只算一步，很可能被狡猾的對手欺騙，所以為了多想幾步，還需要遞歸和回朔。在當時的機器上，算 3 步就基本上需要 3 秒左右的時間了。后來大學畢業收拾東西的時候找到這個程序，試了一下，發現算 10 步需要的時間也基本上感覺不出來了。

　　不知道你是否有同樣的經歷，我們不知不覺的一直在享受著這樣的免費午餐。可是，隨著摩爾定律的提前終結，免費的午餐終究要還回去。雖然硬件設計師還在努力：Hyper Threading CPU（多出一套寄存器，相當于一個邏輯CPU）使得 Pipeline 盡可能滿負荷，使多個 Thread 的操作有可能并行，使得多線程程序的性能有 5%-15% 的提升；增加 Cache 容量也使得包括 Single-Thread 和 Multi-Thread 程序都能受益。也許這些還能幫助你一段時間，但問題是，我們必須做出改變，面對這個即將到來的變革，你準備好了么？

　　Concurrency Programming != Multi-Thread Programming。很多人都會說 MultiThreading 誰不會，問題是，你是為什么使用/如何使用多線程的？我從前做過一個類似 AcdSee 一樣的圖像查看/處理程序，我通常用它來處理我的數碼照片。我在里面用了大量的多線程，不過主要目的是在圖像處理的時候不要 Block 住 UI，所以將 CPU Intensive 的計算部分用后臺線程進行處理，而并沒有把對圖像矩陣的運算并行分開。

　　我覺得 Concurrency Programming 真正的挑戰在于 Programming Model 的改變，在程序員的腦子里面要對自己的程序怎樣并行化有很清楚的認識，更重要的是，如何去實現（包括架構、容錯、實時監控等等）這種并行化，如何去調試，如何去測試。

　　在 Google，每天有海量的數據需要在有限的時間內進行處理（其實每個互聯網公司都會碰到這樣的問題），每個程序員都需要進行分布式的程序開發，這其中包括如何分布、調度、監控以及容錯等等。Google的 MapReduce 正是把分布式的業務邏輯從這些復雜的細節中抽象出來，使得沒有或者很少并行開發經驗的程序員也能進行并行應用程序的開發。

　　MapReduce 中最重要的兩個詞就是Map（映射）和 Reduce（規約）。初看 Map/Reduce 這兩個詞，熟悉 Function Language 的人一定感覺很熟悉。FP 把這樣的函數稱為“higher order function”（“High order function” 被成為 Function Programming 的利器之一哦），也就是說，這些函數是被編寫來與其它函數相結合（或者說被其它函數調用的）。如果說硬要比的化，可以把它想象成 C 里面的 CallBack 函數，或者 STL 里面的 Functor。比如你要對一個 STL 的容器進行查找，需要制定每兩個元素相比較的Functor(Comparator)，這個 Comparator 在遍歷容器的時候就會被調用。

　　拿前面說過圖像處理程序來舉例，其實大多數的圖像處理操作都是對圖像矩陣進行某種運算。這里的運算通常有兩種，一種是映射，一種是規約。拿兩種效果來說，”老照片”效果通常是強化照片的 G/B 值，然后對每個象素加一些隨機的偏移，這些操作在二維矩陣上的每一個元素都是獨立的，是 Map 操作。而”雕刻”效果需要提取圖像邊緣，就需要元素之間的運算了，是一種 Reduce 操作。再舉個簡單的例子，一個一維矩陣（數組）[0,1,2,3,4] 可以映射為 [0,2,3,6,8]（乘2），也可以映射為[1,2,3,4,5]（加1）。它可以規約為0（元素求積）也可以規約為10（元素求和）。

　　面對復雜問題，古人教導我們要“分而治之”，英文中對應的詞是”Divide and Conquer“。Map/Reduce 其實就是 Divide/Conquer 的過程，通過把問題 Divide，使這些Divide 后的 Map 運算高度并行，再將 Map 后的結果 Reduce（根據某一個Key），得到最終的結果。

　　Googler 發現這是問題的核心，其它都是共性問題。因此，他們把 Map/Reduce 抽象分離出來。這樣，Google 的程序員可以只關心應用邏輯，關心根據哪些 Key 把問題進行分解，哪些操作是 Map 操作，哪些操作是 Reduce 操作。其它并行計算中的復雜問題諸如分布、工作調度、容錯、機器間通信都交給 Map/Reduce Framework 去做，很大程度上簡化了整個編程模型。

　　MapReduce 的另一個特點是，Map 和 Reduce 的輸入和輸出都是中間臨時文件（MapReduce 利用 Google 文件系統來管理和訪問這些文件），而不是不同進程間或者不同機器間的其它通信方式。我覺得，這是 Google 一貫的風格，化繁為簡，返璞歸真。

　　接下來就放下其它，研究一下 Map/Reduce 操作。（其它比如容錯、備份任務也有很經典的經驗和實現，論文里面都有詳述）

　　Map的定義：

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.

　　Reduce的定義：

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.

　　MapReduce 論文中給出了這樣一個例子：在一個文檔集合中統計每個單詞出現的次數。

　　Map 操作的輸入是每一篇文檔，將輸入文檔中每一個單詞的出現輸出到中間文件中去。

map(String key, String value):
    // key: document name
    // value: document contents
    for each word w in value:
        EmitIntermediate(w, “1″);

　　比如我們有兩篇文檔，內容分別是

　　A － “I love programming”

　　B － “I am a blogger, you are also a blogger”。

　　B 文檔經過 Map 運算后輸出的中間文件將會是：

I,1
am,1
a,1
blogger,1
you,1
are,1
a,1
blogger,1

　　Reduce 操作的輸入是單詞和出現次數的序列。用上面的例子來說，就是 (“I”, [1, 1]), (“love”, [1]), (“programming”, [1]), (“am”, [1]), (“a”, [1,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));

　　最后輸出的最終結果就會是：(“I”, 2″), (“a”, 2″)……

　　實際的執行順序是：

1. MapReduce Library 將 Input 分成 M 份。這里的 Input Splitter 也可以是多臺機器并行 Split。
2. Master 將 M 份 Job 分給 Idle 狀態的 M 個 worker 來處理；
3. 對于輸入中的每一個 <key, value> pair 進行 Map 操作，將中間結果 Buffer 在 Memory 里；
4. 定期的（或者根據內存狀態），將 Buffer 中的中間信息 Dump 到本地磁盤上，并且把文件信息傳回給 Master（Master 需要把這些信息發送給 Reduce worker）。這里最重要的一點是，在寫磁盤的時候，需要將中間文件做Partition（比如R個）。拿上面的例子來舉例，如果把所有的信息存到一個文件，Reduce worker 又會變成瓶頸。我們只需要保證相同 Ke y能出現在同一個 Partition 里面就可以把這個問題分解。
5. R 個 Reduce worker 開始工作，從不同的 Map worker 的Partition那里拿到數據（read the buffered data from the local disks of the map workers），用key 進行排序（如果內存中放不下需要用到外部排序 – external sort）。很顯然，排序（或者說Group）是 Reduce 函數之前必須做的一步。 這里面很關鍵的是，每個Reduce worker 會去從很多 Map worker 那里拿到 X(0<X<R) Partition 的中間結果，這樣，所有屬于這個 Key 的信息已經都在這個 worker 上了。
6. Reduce worker 遍歷中間數據，對每一個唯一 Key，執行 Reduce 函數（參數是這個 key 以及相對應的一系列 Value）。
7. 執行完畢后，喚醒用戶程序，返回結果（最后應該有 R 份 Output，每個 Reduce Worker 一個）。

　　可見，這里的分（Divide）體現在兩步，分別是將輸入分成 M 份，以及將 Map 的中間結果分成R份。將輸入分開通常很簡單，Map 的中間結果通常用“hash(key) mod R”這個結果作為標準，保證相同的 Key 出現在同一個 Partition 里面。當然，使用者也可以指定自己的 Partition Function，比如，對于 Url Key，如果希望同一個 Host 的 URL 出現在同一個 Partition，可以用“hash(Hostname(urlkey)) mod R”作為 Partition Function。

　　對于上面的例子來說，每個文檔中都可能會出現成千上萬的 (“the”, 1)這樣的中間結果，瑣碎的中間文件必然導致傳輸上的損失。因此，MapReduce還 支持用戶提供 Combiner Function。這個函數通常與 Reduce Function 有相同的實現，不同點在于 Reduce 函數的輸出是最終結果，而 Combiner 函數的輸出是 Reduce 函數的某一個輸入的中間文件。

　　Tom White給出了 Nutch[2] 中另一個很直觀的例子，分布式Grep。我一直覺得，Pipe 中的很多操作，比如 More、Grep、Cat 都類似于一種 Map 操作，而 Sort、Uniq、wc 等都相當于某種 Reduce 操作。

　　加上前兩天 Google 剛剛發布的 BigTable 論文，現在 Google 有了自己的集群 – Googel Cluster，分布式文件系統 – GFS，分布式計算環境 – MapReduce，分布式結構化存儲 – BigTable，再加上 Lock Service。我真的能感覺的到 Google 著名的免費晚餐之外的對于程序員的另一種免費的晚餐，那個由大量的 commodity PC 組成的 large clusters。我覺得這些才真正是Google的核心價值所在。

呵　　呵，就像微軟老兵 Joel Spolsky（你應該看過他的”Joel on Software”吧？）曾經說過，對于微軟來說最可怕的是[1]，微軟還在苦苦追趕 Google 來完善 Search 功能的時候，Google 已經在部署下一代的超級計算機了。

The very fact that Google invented MapReduce, and Microsoft didn’t, says something about why Microsoft is still playing catch up trying to get basic search features to work, while Google has moved on to the next problem: building Skynet^H^H^H^H^H^H the world’s largest massively parallel supercomputer. I don’t think Microsoft completely understands just how far behind they are on that wave.

　　注1：其實，微軟也有自己的方案 – DryAd。問題是，大公司里，要想重新部署這樣一個底層的 InfraStructure，無論是技術的原因，還是政治的原因，將是如何的難。

　　注2：Lucene 之父Doug Cutting的又一力作，Project Hadoop - 由Hadoop分布式文件系統和一個 Map/Reduce 的實現組成，Lucene/Nutch 的成產線也夠齊全的了。