Ch.2.MapReduce簡介南京大學計算機科學與技術系主講人：黃宜華2011年春季學期MapReduce海量數據并行處理鳴謝：本課程得到Google公司(北京）中國大學合作部精品課程計劃資助Ch.2.

MapReduce簡介1.對付大數據處理-分而治之2.構建抽象模型-Map和Reduce3.上升到構架-自動并行化并隱藏低層細節4.MapReduce的主要設計思想和特征大規模數據處理時，MapReduce在三個層面上的基本構思如何對付大數據處理：分而治之

對相互間不具有計算依賴關系的大數據，實現并行最自然的辦法就是采取分而治之的策略上升到抽象模型：Mapper與Reducer

MPI等并行計算方法缺少高層并行編程模型，為了克服這一缺陷，MapReduce借鑒了Lisp函數式語言中的思想，用Map和Reduce兩個函數提供了高層的并行編程抽象模型上升到構架：統一構架，為程序員隱藏系統層細節

MPI等并行計算方法缺少統一的計算框架支持，程序員需要考慮數據存儲、劃分、分發、結果收集、錯誤恢復等諸多細節；為此，MapReduce設計并提供了統一的計算框架，為程序員隱藏了絕大多數系統層面的處理細節什么樣的計算任務可進行并行化計算？

并行計算的第一個重要問題是如何劃分計算任務或者計算數據以便對劃分的子任務或數據塊同時進行計算。

但一些計算問題恰恰無法進行這樣的劃分！

Ninewomencannothaveababyinonemonth!例如：Fibonacci函數:Fk+2=Fk+Fk+1

前后數據項之間存在很強的依賴關系！只能串行計算！

結論：不可分拆的計算任務或相互間有依賴關系的數據無法進行并行計算！1.如何對付大數據處理：分而治之大數據的并行化計算一個大數據若可以分為具有同樣計算過程的數據塊，并且這些數據塊之間不存在數據依賴關系，則提高處理速度的最好辦法就是并行計算例如：假設有一個巨大的2維數據需要處理(比如求每個元素的開立方)，其中對每個元素的處理是相同的,并且數據元素間不存在數據依賴關系,可以考慮不同的劃分方法將其劃分為子數組,由一組處理器并行處理

如何對付大數據處理：分而治之大數據的并行化計算

如何對付大數據處理：分而治之合并Master：負責劃分和分配任務Workder：負責數據塊計算大數據任務劃分和并行計算模型

如何對付大數據處理：分而治之大數據計算任務子任務子任務子任務子任務……任務劃分計算結果結果合并借鑒函數式設計語言Lisp的設計思想函數式程序設計(functionalprogramming)語言Lisp是一種列表處理語言(Listprocessing)，是一種應用于人工智能處理的符號式語言，由MIT的人工智能專家、圖靈獎獲得者JohnMcCarthy于1958年設計發明。Lisp定義了可對列表元素進行整體處理的各種操作，如：

如：(add#(1234)#(4321))將產生結果：

#(5555)Lisp中也提供了類似于Map和Reduce的操作

如:(map‘vector#+#(12345)#(1011121314))

通過定義加法map運算將2個向量相加產生結果#(1113151719)

(reduce#’+#(1113151719))通過加法歸并產生累加結果75

2.構建抽象模型：Map與ReduceMap:對一組數據元素進行某種重復式的處理Reduce:對Map的中間結果進行某種進一步的結果整理MPI中的數據規約操作ReduceMPI規約操作編程示例—計算積分(參見Ch.1)

for(i=myid;i<N;i=i+numprocs)/*根據節點數目將N個矩形分為圖示的多個顏色組*/

{/*每個節點計算一個顏色組的矩形面積并累加*/

x=a+i*dx+dx/2;/*以每個矩形的中心點x值計算矩形高度*/local+=x*x*dx;/*矩形面積=高度x寬度=y*dx*/}

MPI_Reduce(&local,&inte,1,MPI_DOUBLE,MPI_SUM,0,MPI_COMM_WORLD);

if(myid==0)/*規約所有節點上的累加和并送到主節點0*/

{/*主節點打印累加和*/

printf("Theintegalofx*xinregion[%d,%d]=%16.15f\n",a,b,inte);

}

MPI_Finalize();

}Theintegal

ofx*xinregion[0,10]=33.33345構建抽象模型：Map與Reduce構建抽象模型：Map與ReduceMPI中的數據規約操作Reduce

將一組進程的數據按照指定的操作方式規約到一起并傳送給一個進程

MPI_Reduce(sendbuf,recvbuf,count,datatype,op,root,comm)其中規約操作op可設為下表定義的操作之一：MPI_MAX 求最大值 MPI_MIN 求最小值MPI_SUM 求和 MPI_PROD 求積MPI_LAND 邏輯與 MPI_BAND 按位與MPI_LOR 邏輯或 MPI_BOR 按位或MPI_LXOR 邏輯異或 MPI_BXOR 按位異或MPI_MAXLOC最大值和位置 MPI_MINLOC 最小值和位置不足：僅能處理以上規定的規約操作，

不能實現靈活復雜的規約操作！構建抽象模型：Map與Reduce

關系數據庫中的聚合函數

對一個查詢操作的結果列表中的字段表達式進行聚合操作

selectOrder_ID,Payment=SUM(Price*Quantity)groupbyOrder_ID

Sum() 計算表達式所有值之和 Avg() 計算表達式的平均值 Count(*) 計算某字段中所有值的個數 Min() 計算表達式的最小值 Max() 計算表達式的最大值Order_IDItemPriceQuantity1電腦500021打印機400011硬盤80032電腦600012硬盤6002查詢結果：Orde_IDPayment

116400

(5000*2+4000*1+800*3)

27200(6000*1+600*2)數據庫中的這些聚合函數類似于對表格數據進行的Reduce操作典型的流式大數據問題的特征大量數據記錄/元素進行重復處理對每個數據記錄/元素作感興趣的處理、獲取感興趣的中間結果信息排序和整理中間結果以利后續處理收集整理中間結果產生最終結果輸出構建抽象模型：Map與ReduceMapReduce關鍵思想：為大數據處理過程中的兩個主要處理操作

提供一種抽象機制MapReduce中的Map和Reduce操作的抽象描述

MapReduce借鑒了函數式程序設計語言Lisp中的思想，定義了如下的Map和Reduce兩個抽象的編程接口，由用戶去編程實現:map:(k1;v1)

[(k2;v2)]輸入：鍵值對(k1;v1)表示的數據處理：文檔數據記錄(如文本文件中的行，或數據表格中的行)將以“鍵值對”形式傳入map函數；map函數將處理這些鍵值對，并以另一種鍵值對形式輸出處理的一組鍵值對中間結果[(k2;v2)]輸出：鍵值對[(k2;v2)]表示的一組中間數據構建抽象模型：Map與ReduceMapReduce中的Map和Reduce操作的抽象描述

reduce:(k2;[v2])

[(k3;v3)]輸入：

由map輸出的一組鍵值對[(k2;v2)]將被進行合并處理將同樣主鍵下的不同數值合并到一個列表[v2]中，故reduce的輸入為(k2;[v2])處理：對傳入的中間結果列表數據進行某種整理或進一步的處理,并產生最終的某種形式的結果輸出[(k3;v3)]。輸出：最終輸出結果[(k3;v3)]Map和Reduce為程序員提供了一個清晰的操作接口抽象描述構建抽象模型：Map與Reduce基于Map和Reduce的并行計算模型構建抽象模型：Map與Reduce海量數據存儲……數據劃分MapMapMapMap初始kv鍵值對初始kv鍵值對初始kv鍵值對初始kv鍵值對中間結果(k1,val)(k2,val)(k3,val)(k1,val)(k3,val)(k2,val)(k3,val)(k1,val)(k2,val)(k3,val)Barrier：AggregationandShuffleReduceReduceReduce(k1,values)(k2,values)(k3,values)計算結果(K1,val)(K2,val)(K3,val)基于Map和Reduce的并行計算模型各個map函數對所劃分的數據并行處理，從不同的輸入數據產生不同的中間結果輸出各個reduce也各自并行計算，各自負責處理不同的中間結果數據集合進行reduce處理之前,必須等到所有的map函數做完，因此,在進入reduce前需要有一個同步障(barrier);這個階段也負責對map的中間結果數據進行收集整理(aggregation&shuffle)處理,以便reduce更有效地計算最終結果最終匯總所有reduce的輸出結果即可獲得最終結果構建抽象模型：Map與Reduce基于MapReduce的處理過程示例--文檔詞頻統計：WordCount設有4組原始文本數據：Text1:theweatherisgoodText2:todayisgoodText3:goodweatherisgoodText4:today

hasgoodweather傳統的串行處理方式(Java)：

String[]text=newString[]{“helloworld”,“helloeveryone”,“sayhellotoeveryoneintheworld”｝;HashTableht=newHashTable();for(i=0;i<3;++i){StringTokenizerst=newStringTokenizer(text[i]);while(st.hasMoreTokens()){Stringword=st.nextToken();if(!ht.containsKey(word)){ht.put(word,newInteger(1));}else{intwc=((Integer)ht.get(word)).intValue()+1;//計數加1ht.put(word,newInteger(wc));}}}for(Iteratoritr=ht.KeySet().iterator();itr.hasNext();){Stringword=(String)itr.next();System.out.print(word+“:”+(Integer)ht.get(word)+“;”);}構建抽象模型：Map與Reduce輸出：good:5;has:1;is:3;the:1;today:2;weather:3基于MapReduce的處理過程示例--文檔詞頻統計：WordCountMapReduce處理方式使用4個map節點：map節點1:

輸入：(text1,“theweatherisgood”)

輸出：(the,1),(weather,1),(is,1),(good,1)map節點2:

輸入：(text2,“todayisgood”)

輸出：(today,1),(is,1),(good,1)map節點3:

輸入：(text3,“goodweatherisgood”)

輸出：(good,1),(weather,1),(is,1),(good,1)map節點4:

輸入：(text3,“todayhasgoodweather”)

輸出：(today,1),(has,1),(good,1),(weather,1)構建抽象模型：Map與Reduce基于MapReduce的處理過程示例--文檔詞頻統計：WordCountMapReduce處理方式使用3個reduce節點：reduce節點1:

輸入：(good,1),(good,1),(good,1),(good,1),(good,1)

輸出：(good,5)reduce節點2:

輸入：(has,1),(is,1),(is,1),(is,1),

輸出：(has,1),(is,3)reduce節點3:

輸入：(the,1),(today,1),(today,1)(weather,1),(weather,1),(weather,1)

輸出：(the,1),(today,2),(weather,3)構建抽象模型：Map與Reduce輸出：good:5is:3has:1the:1today:2weather:3基于MapReduce的處理過程示例--文檔詞頻統計：WordCountMapReduce處理方式MapReduce偽代碼(實現Map和Reduce兩個函數)：構建抽象模型：Map與ReduceClassMappermethodmap(Stringinput_key,Stringinput_value):

//input_key:textdocumentname//input_value:documentcontents

foreachwordwininput_value:

EmitIntermediate(w,"1");ClassReducermethodreduce(Stringoutput_key,Iteratorintermediate_values):

//output_key:aword//output_values:alistofcounts

intresult=0;

foreachvinintermediate_values:result+=ParseInt(v);

Emit(AsString(result));如何提供統一的計算框架主要需求和目標：實現自動并行化計算為程序員隱藏系統層細節需要考慮的細節技術問題：如何管理和存儲數據？如何劃分數據？如何調度計算任務并分配map和reduce節點？如果節點間需要共享或交換數據怎么辦？如何考慮數據通信和同步？如何掌控節點的執行完成情況？如何收集中間和最終的結果數據？節點失效如何處理？如何恢復數據？如何恢復計算任務？節點擴充后如何保證原有程序仍能正常運行并保證系統性能提升？問題：我們能把這些細節和復雜性交給系統去負責處理嗎？3.上升到構架:自動并行化并隱藏底層細節如何提供統一的計算框架答案：MapReduce之前的并行計算方法都未能做到

但MapReduce做到了！MapReduce提供一個統一的計算框架，可完成：計算任務的劃分和調度數據的分布存儲和劃分處理數據與計算任務的同步結果數據的收集整理(sorting,combining,partitioning,…)系統通信、負載平衡、計算性能優化處理處理系統節點出錯檢測和失效恢復上升到構架：自動化并行并隱藏低層細節如何提供統一的計算框架MapReduce最大的亮點通過抽象模型和計算框架把需要做什么(whatneedtodo)與具體怎么做(howtodo)分開了，為程序員提供一個抽象和高層的編程接口和框架程序員僅需要關心其應用層的具體計算問題，僅需編寫少量的處理應用本身計算問題的程序代碼如何具體完成這個并行計算任務所相關的諸多系統層細節被隱藏起來,交給計算框架去處理：從分布代碼的執行，到大到數千小到單個節點集群的自動調度使用上升到構架：自動化并行并隱藏低層細節如何提供統一的計算框架MapReduce提供的主要功能*任務調度：提交的一個計算作業(job)將被劃分為很多個計算任務(tasks),任務調度功能主要負責為這些劃分后的計算任務分配和調度計算節點(map節點或reducer節點);同時負責監控這些節點的執行狀態,并負責map節點執行的同步控制(barrier);也負責進行一些計算性能優化處理,如對最慢的計算任務采用多備份執行、選最快完成者作為結果數據/代碼互定位：為了減少數據通信，一個基本原則是本地化數據處理(locality)，即一個計算節點盡可能處理其本地磁盤上所分布存儲的數據，這實現了代碼向數據的遷移；當無法進行這種本地化數據處理時，再尋找其它可用節點并將數據從網絡上傳送給該節點(數據向代碼遷移)，但將盡可能從數據所在的本地機架上尋找可用節點以減少通信延遲上升到構架：自動化并行并隱藏低層細節*CitefromJimmyLin,University

ofMaryland,Data-IntensiveTextprocessingwithMapReduce如何提供統一的計算框架MapReduce提供的主要功能出錯處理：以低端商用服務器構成的大規模MapReduce計算集群中,節點硬件(主機、磁盤、內存等)出錯和軟件有bug是常態，因此,MapReducer需要能檢測并隔離出錯節點，并調度分配新的節點接管出錯節點的計算任務分布式數據存儲與文件管理：海量數據處理需要一個良好的分布數據存儲和文件管理系統支撐,該文件系統能夠把海量數據分布存儲在各個節點的本地磁盤上,但保持整個數據在邏輯上成為一個完整的數據文件；為了提供數據存儲容錯機制,該文件系統還要提供數據塊的多備份存儲管理能力Combiner和Partitioner:為了減少數據通信開銷,中間結果數據進入reduce節點前需要進行合并(combine)處理,把具有同樣主鍵的數據合并到一起避免重復傳送;一個reducer節點所處理的數據可能會來自多個map節點,因此,map節點輸出的中間結果需使用一定的策略進行適當的劃分(partitioner)處理，保證相關數據發送到同一個reducer節點上升到構架：自動化并行并隱藏低層細節Barrier(good,1)(good,1)(good,2)(good,1)PartitionerPartitionerPartitionerPartitioner(is,1)(is,1)(is,1)(has,1)(weather,1)(weather,1)(weather,1)(the,1)(today,1)(today,1)基于Map和Reduce的并行計算模型構建抽象模型：Map與Reduce海量數據存儲計算結果……數據劃分Map初始kv鍵值對初始kv鍵值對初始kv鍵值對初始kv鍵值對MapMapMap中間結果(the,1)(weather,1)(is,1)(good,1)CombinerCombinerCombinerCombiner(the,1)(weather,1)(is,1)(good,1)(today,1)(is,1)(good,1)(good,1)(weather,1)(is,1)(good,1)(today,1)(has,1)(good,1)(weather,1)(today,1)(is,1)(good,1)(good,2)(weather,1)(is,1)(today,1)(has,1)(good,1)(weather,1)ReduceReduceReduce(good,5)(is,3)(has,1)(weather,3)(the,1)(today,2)Combiner和Partitioner4.MapReduce的主要設計思想與特點*

向“外”橫向擴展，而非向“上”縱向擴展

Scale“out",not“up”

即MapReduce集群的構筑選用價格便宜、易于擴展的大量低端商用服務器，而非價格昂貴、不易擴展的高端服務器（SMP）低端服務器市場與高容量DesktopPC有重疊的市場，因此，由于相互間價格的競爭、可互換的部件、和規模經濟效應，使得低端服務器保持較低的價格基于TPC-C在2007年低的性能評估結果,一個低端服務器平臺與高端的共享存儲器結構的服務器平臺相比,其性價比大約要高4倍;如果把外存價格除外,低端服務器性價比大約提高12倍對于大規模數據處理，由于有大量數據存儲需要，顯而易見，基于低端服務器的集群遠比基于高端服務器的集群優越，這就是為什么MapReduce并行計算集群會基于低端服務器實現*CitefromJimmyLin,University

ofMaryland,Data-IntensiveTextprocessingwithMapReduceMapReduce的主要設計思想與特點

失效被認為是常態

AssumefailuresarecommonMapReduce集群中使用大量的低端服務器(Google目前在全球共使用百萬臺以上的服務器節點),因此，節點硬件失效和軟件出錯是常態，因而：一個良好設計、具有容錯性的并行計算系統不能因為節點失效而影響計算服務的質量，任何節點失效都不應當導致結果的不一致或不確定性；任何一個節點失效時，其它節點要能夠無縫接管失效節點的計算任務；當失效節點恢復后應能自動無縫加入集群，而不需要管理員人工進行系統配置MapReduce并行計算軟件框架使用了多種有效的機制，如節點自動重啟技術，使集群和計算框架具有對付節點失效的健壯性，能有效處理失效節點的檢測和恢復。

把處理向數據遷移

Movingprocessingtothedata傳統高性能計算系統通常有很多處理器節點與一些外存儲器節點相連，如用區域存儲網絡(SAN,StorageAreaNetwork)連接的磁盤陣列，因此，大規模數據處理時外存文件數據I/O訪問會成為一個制約系統性能的瓶頸。為了減少大規模數據并行計算系統中的數據通信開銷，代之以把數據傳送到處理節點(數據向處理器或代碼遷移)，應當考慮將處理向數據靠攏和遷移。MapReduce采用了數據/代碼互定位的技術方法，計算節點將首先將盡量負責計算其本地存儲的數據,以發揮數據本地化特點(locality),僅當節點無法處理本地數據時，再采用就近原則尋找其它可用計算節點，并把數據傳送到該可用計算節點。MapReduce的主要設計思想與特點

順序處理數據、避免隨機訪問數據

Processdatasequentiallyandavoidrandomaccess大規模數據處理的特點決定了大量的數據記錄不可能存放在內存、而只可能放在外存中進行處理。磁盤的順序訪問和隨即訪問在性能上有巨大的差異

例：100億(1010)個數據記錄(每記錄100B,共計1TB)的數據庫

更新1%的記錄(一定是隨機訪問)需要1個月時間；

而順序訪問并重寫所有數據記錄僅需1天時間！MapReduce設計為面向大數據集批處理的并行計算系統，所有計算都被組織成很長的流式操作，以便能利用分布在集群中大量節點上磁盤集合的高傳輸帶寬。MapReduce的主要設計思想與特點

為應用開發者隱藏系統層細節

Hidesystem-leveldetailsfromtheapplicationdeveloper軟件工程實踐指南中，專業程序員認為之所以寫程序困難，是因為程序員需要記住太多的編程細節(從變量