MapReduce海量數(shù)據(jù)并行處理ch.02

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

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

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

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

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

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

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

Ninewomencannothaveababyinonemonth!例如：Fibonacci函數(shù):Fk+2=Fk+Fk+1

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

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

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

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

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

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

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

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

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

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

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

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

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

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)/*規(guī)約所有節(jié)點上的累加和并送到主節(jié)點0*/

{/*主節(jié)點打印累加和*/

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

}

MPI_Finalize();

}Theintegal

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

將一組進程的數(shù)據(jù)按照指定的操作方式規(guī)約到一起并傳送給一個進程

MPI_Reduce(sendbuf,recvbuf,count,datatype,op,root,comm)其中規(guī)約操作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 最小值和位置不足：僅能處理以上規(guī)定的規(guī)約操作，

不能實現(xiàn)靈活復雜的規(guī)約操作！構建抽象模型：Map與Reduce

關系數(shù)據(jù)庫中的聚合函數(shù)

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

selectOrder_ID,Payment=SUM(Price*Quantity)groupbyOrder_ID

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

116400

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

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

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

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

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

reduce:(k2;[v2])

[(k3;v3)]輸入：

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

hasgoodweather傳統(tǒng)的串行處理方式(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;//計數(shù)加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的處理過程示例--文檔詞頻統(tǒng)計：WordCountMapReduce處理方式使用4個map節(jié)點：map節(jié)點1:

輸入：(text1,“theweatherisgood”)

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

輸入：(text2,“todayisgood”)

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

輸入：(text3,“goodweatherisgood”)

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

輸入：(text3,“todayhasgoodweather”)

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

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

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

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

輸出：(has,1),(is,3)reduce節(jié)點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的處理過程示例--文檔詞頻統(tǒng)計：WordCountMapReduce處理方式MapReduce偽代碼(實現(xiàn)Map和Reduce兩個函數(shù))：構建抽象模型：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));如何提供統(tǒng)一的計算框架主要需求和目標：實現(xiàn)自動并行化計算為程序員隱藏系統(tǒng)層細節(jié)需要考慮的細節(jié)技術問題：如何管理和存儲數(shù)據(jù)？如何劃分數(shù)據(jù)？如何調度計算任務并分配map和reduce節(jié)點？如果節(jié)點間需要共享或交換數(shù)據(jù)怎么辦？如何考慮數(shù)據(jù)通信和同步？如何掌控節(jié)點的執(zhí)行完成情況？如何收集中間和最終的結果數(shù)據(jù)？節(jié)點失效如何處理？如何恢復數(shù)據(jù)？如何恢復計算任務？節(jié)點擴充后如何保證原有程序仍能正常運行并保證系統(tǒng)性能提升？問題：我們能把這些細節(jié)和復雜性交給系統(tǒng)去負責處理嗎？3.上升到構架:自動并行化并隱藏底層細節(jié)如何提供統(tǒng)一的計算框架答案：MapReduce之前的并行計算方法都未能做到

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

ofMaryland,Data-IntensiveTextprocessingwithMapReduce如何提供統(tǒng)一的計算框架MapReduce提供的主要功能出錯處理：以低端商用服務器構成的大規(guī)模MapReduce計算集群中,節(jié)點硬件(主機、磁盤、內存等)出錯和軟件有bug是常態(tài)，因此,MapReducer需要能檢測并隔離出錯節(jié)點，并調度分配新的節(jié)點接管出錯節(jié)點的計算任務分布式數(shù)據(jù)存儲與文件管理：海量數(shù)據(jù)處理需要一個良好的分布數(shù)據(jù)存儲和文件管理系統(tǒng)支撐,該文件系統(tǒng)能夠把海量數(shù)據(jù)分布存儲在各個節(jié)點的本地磁盤上,但保持整個數(shù)據(jù)在邏輯上成為一個完整的數(shù)據(jù)文件；為了提供數(shù)據(jù)存儲容錯機制,該文件系統(tǒng)還要提供數(shù)據(jù)塊的多備份存儲管理能力Combiner和Partitioner:為了減少數(shù)據(jù)通信開銷,中間結果數(shù)據(jù)進入reduce節(jié)點前需要進行合并(combine)處理,把具有同樣主鍵的數(shù)據(jù)合并到一起避免重復傳送;一個reducer節(jié)點所處理的數(shù)據(jù)可能會來自多個map節(jié)點,因此,map節(jié)點輸出的中間結果需使用一定的策略進行適當?shù)膭澐?partitioner)處理，保證相關數(shù)據(jù)發(fā)送到同一個reducer節(jié)點上升到構架：自動化并行并隱藏低層細節(jié)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海量數(shù)據(jù)存儲計算結果……數(shù)據(jù)劃分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有重疊的市場，因此，由于相互間價格的競爭、可互換的部件、和規(guī)模經濟效應，使得低端服務器保持較低的價格基于TPC-C在2007年低的性能評估結果,一個低端服務器平臺與高端的共享存儲器結構的服務器平臺相比,其性價比大約要高4倍;如果把外存價格除外,低端服務器性價比大約提高12倍對于大規(guī)模數(shù)據(jù)處理，由于有大量數(shù)據(jù)存儲需要，顯而易見，基于低端服務器的集群遠比基于高端服務器的集群優(yōu)越，這就是為什么MapReduce并行計算集群會基于低端服務器實現(xiàn)*CitefromJimmyLin,University

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

失效被認為是常態(tài)

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

把處理向數(shù)據(jù)遷移

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

順序處理數(shù)據(jù)、避免隨機訪問數(shù)據(jù)

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

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

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

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

為應用開發(fā)者隱藏系統(tǒng)層細節(jié)

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