第3章 并行計算技術

第3章并行計算技術提綱并行計算技術概述分布式數(shù)據(jù)處理MapReducehadoopMR技術的簡單應用單核處理器性能提升接近極限！集成度性能為什么需要并行計算？世界太復雜、單機Handle不了大數(shù)據(jù)動則以TB記的數(shù)據(jù)，信息爆炸時代的數(shù)據(jù)量遠遠不只是TB級別2006年，Google透露存儲了12000PB的數(shù)據(jù)物聯(lián)網(wǎng)越來越多的設備接入互聯(lián)網(wǎng)，產(chǎn)生更多的數(shù)據(jù)并行計算!!!SMPMPPClusterGRIDCloudMulticoreManycore解決方案?不可并行的任務例1：

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

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

結論：不可分拆的計算任務或相互間有依賴關系的數(shù)據(jù)無法進行并行計算！分布式數(shù)據(jù)處理MapReduce

產(chǎn)生背景

編程模型實現(xiàn)機制案例分析如何對付大數(shù)據(jù)處理：分而治之大數(shù)據(jù)的并行化計算一個大數(shù)據(jù)若可以分為具有同樣計算過程的數(shù)據(jù)塊，并且這些數(shù)據(jù)塊之間不存在數(shù)據(jù)依賴關系，則提高處理速度的最好辦法就是并行計算例如：假設有一個巨大的2維數(shù)據(jù)需要處理(比如求每個元素的開立方)，其中對每個元素的處理是相同的,并且數(shù)據(jù)元素間不存在數(shù)據(jù)依賴關系,可以考慮不同的劃分方法將其劃分為子數(shù)組,由一組處理器并行處理

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

合并Master：負責劃分和分配任務Workder：負責數(shù)據(jù)塊計算大數(shù)據(jù)任務劃分和并行計算模型大數(shù)據(jù)計算任務子任務子任務子任務子任務……任務劃分計算結果結果合并MapReduce在三個層面上的基本構思分而治之對付大數(shù)據(jù)處理

對相互間不具有計算依賴關系的大數(shù)據(jù)，實現(xiàn)并行最自然的辦法就是采取分而治之的策略Mapper與ReducerMapReduce借鑒了Lisp函數(shù)式語言中的思想，用Map和Reduce兩個函數(shù)提供了高層的并行編程抽象模型統(tǒng)一構架，為程序員隱藏系統(tǒng)層細節(jié)MapReduce設計并提供了統(tǒng)一的計算框架，為程序員隱藏了絕大多數(shù)系統(tǒng)層面的處理細節(jié)海量數(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的并行計算模型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ù)

reduce:(k2;[v2])

[(k3;v3)]輸入：

由map輸出的一組鍵值對[(k2;v2)]將被進行合并處理將同樣主鍵下的不同數(shù)值合并到一個列表[v2]中，故reduce的輸入為(k2;[v2])處理：對傳入的中間結果列表數(shù)據(jù)進行某種整理或進一步的處理,并產(chǎn)生最終的某種形式的結果輸出[(k3;v3)]。輸出：最終輸出結果[(k3;v3)]Map和Reduce為程序員提供了一個清晰的操作接口抽象描述基于Map和Reduce的并行計算模型各個map函數(shù)對所劃分的數(shù)據(jù)并行處理，從不同的輸入數(shù)據(jù)產(chǎn)生不同的中間結果輸出各個reduce也各自并行計算，各自負責處理不同的中間結果數(shù)據(jù)集合進行reduce處理之前,必須等到所有的map函數(shù)做完，因此,在進入reduce前需要有一個同步障(barrier);這個階段也負責對map的中間結果數(shù)據(jù)進行收集整理(aggregation&shuffle)處理,以便reduce更有效地計算最終結果最終匯總所有reduce的輸出結果即可獲得最終結果2023/1/31基于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)+“;”);}輸出：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)基于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)輸出：good:5is:3has:1the:1today:2weather:3MapReduce偽代碼(實現(xiàn)Map和Reduce兩個函數(shù))：ClassMappermethodmap(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(output_key，result);MapReduce提供統(tǒng)一的計算框架主要需求和目標：實現(xiàn)自動并行化計算為程序員隱藏系統(tǒng)層細節(jié)MapReduce之前的并行計算方法都未能做到

但MapReduce做到了！MapReduce提供一個統(tǒng)一的計算框架：計算任務的劃分和調(diào)度數(shù)據(jù)的分布存儲和劃分處理數(shù)據(jù)與計算任務的同步結果數(shù)據(jù)的收集整理(sorting,combining,partitioning,…)系統(tǒng)通信、負載平衡、計算性能優(yōu)化處理處理系統(tǒng)節(jié)點出錯檢測和失效恢復MapReduce最大的亮點通過抽象模型和計算框架把需要做什么(whatneedtodo)與具體怎么做(howtodo)分開了，為程序員提供一個抽象和高層的編程接口和框架程序員僅需要關心其應用層的具體計算問題，僅需編寫少量的處理應用本身計算問題的程序代碼如何具體完成這個并行計算任務所相關的諸多系統(tǒng)層細節(jié)被隱藏起來,交給計算框架去處理：從分布代碼的執(zhí)行，到大到數(shù)千小到單個節(jié)點集群的自動調(diào)度使用MapReduce提供的主要功能*任務調(diào)度：提交的一個計算作業(yè)(job)將被劃分為很多個計算任務(tasks),任務調(diào)度功能主要負責為這些劃分后的計算任務分配和調(diào)度計算節(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ù)從網(wǎng)絡上傳送給該節(jié)點(數(shù)據(jù)向代碼遷移)，但將盡可能從數(shù)據(jù)所在的本地機架上尋找可用節(jié)點以減少通信延遲出錯處理：以低端商用服務器構成的大規(guī)模MapReduce計算集群中,節(jié)點硬件(主機、磁盤、內(nèi)存等)出錯和軟件有bug是常態(tài)，因此,MapReducer需要能檢測并隔離出錯節(jié)點，并調(diào)度分配新的節(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é)點*CitefromJimmyLin,University

ofMaryland,Data-IntensiveTextprocessingwithMapReduceBarrier(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)海量數(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和Partitioner基于Map和Reduce的并行計算模型平滑無縫的可擴展性

CitefromAndréaMatsunagaet.al.CloudBLAST:CombiningMapReduceandVirtualizationonDistributedResourcesforBioinformaticsApplications.2008基于MapReduce的基因序列比對算法BLAST的研究顯示,無論基于虛擬機還是非虛擬機MapReduce，隨著處理器數(shù)目的增加都能實現(xiàn)近似于線性的性能增長Hadoop簡介

Hadoop云計算系統(tǒng)Google云計算系統(tǒng)HadoopHDFSGoogleGFSHadoopMapReduceGoogleMapReduceHadoopHBaseGoogleBigtableHadoopZooKeeperGoogleChubbyHadoopPigGoogleSawzallHadoop——Apache開源組織的一個分布式計算框架，可以在大量廉價的硬件設備組成的集群上運行應用程序，為應用程序提供了一組穩(wěn)定可靠的接口，旨在構建一個具有高可靠性和良好擴展性的分布式系統(tǒng)

Hadoop云計算系統(tǒng)與Google云計算系統(tǒng)

HadoopMapReduce框架2023/1/31masterSlavesHadoopMap/Reduce(input)<k1,v1>->

map

-><k2,v2>->

combine*

-><k2,v2>->reduce

-><k3,v3>(output)combine過程可能沒有，也可能有多次InjectPointsInputjob.setInputFormatClass()SplitRecordReaderMapjob.setMapperClass()Combine*job.setCombinerClass()Shufflingjob.setPartitionerClass()Sortjob.setSortComparatorClass()Groupingjob.setGroupingComparatorClass()Reducejob.setReducerClass()Outputjob.setOutputFormatClass()JobTracker&TaskTrackerJob&Task一個Job會被分成多個Task執(zhí)行一個Task對應一個Map或者ReduceJobTracker運行在Master上，管理和跟蹤每個Job收集Task的信息狀態(tài)，并匯總重新調(diào)度失敗的任務TaskTracker向JobTracker匯報狀態(tài)（心跳）運行在每個計算節(jié)點上，管理和跟蹤每個Task收集task的信息，并提供給JobTrackerJobShedulerFIFO先到先得，排隊執(zhí)行FairScheduler（公平調(diào)度器）它的目的是讓所有的作業(yè)隨著時間的推移，都能平均的獲取等同的共享資源。按資源池（pool）來組織作業(yè)，并把資源按配置分到這些資源池里/common/docs/r0.20.2/fair_scheduler.htmlCapacityScheduler（容量調(diào)度器）支持多個queue，每個Job提交到一個queue里支持內(nèi)存調(diào)度，對于需要高內(nèi)存的任務，調(diào)度到有足夠內(nèi)存的節(jié)點/common/docs/r0.20.2/capacity_scheduler.htmlJobSchedulerWebUIFailure當一個Task某次運行失敗后，會選擇其他的計算節(jié)點重新執(zhí)行，直到達到設置的最大失敗次數(shù)如果某個task達到最大失敗次數(shù)時，會觸發(fā)Job失敗JobShellhadoopjob–kill<job-id>hadoopjob–listhadoopjob-set-priority<job-id><priority>hadoopjob-status<job-id>hadoopjob-kill-task<task-id>hadoopjob-fail-task<task-id>WebUI代碼實例

publicstaticvoidmain(String[]args)throwsException{ JobConfconf=newJobConf(WordCount.class); conf.setJobName("wordcount");

conf.setOutputKeyClass(Text.class); conf.setOutputValueClass(IntWritable.class);

conf.setMapperClass(Map.class); conf.setCombinerClass(Reduce.class); conf.setReducerClass(Reduce.class);

conf.setInputFormat(TextInputFormat.class); conf.setOutputFormat(TextOutputFormat.class);

FileInputFormat.setInputPaths(conf,newPath(args[0])); FileOutputFormat.setOutputPath(conf,newPath(args[1]));

JobClient.runJob(conf); }2023/1/31MapClass

publicclassMapextendsMapReduceBaseimplementsMapper<LongWrit