cmcdh集群管理課程培訓(xùn)

1、大數(shù)據(jù)課程之MapReduce(2) 培訓(xùn)講師：吳超 課程安排標(biāo)準(zhǔn)和自定義計(jì)數(shù)器*Combiner和Partitioner編程*自定義排序和分組編程*表連接*常見(jiàn)的調(diào)度器*MR優(yōu)化方法*常見(jiàn)的MapReduce算法*-加深拓展-常見(jiàn)大數(shù)據(jù)處理方法*了解計(jì)數(shù)器hadoop計(jì)數(shù)器:可以讓開(kāi)發(fā)人員以全局的視角來(lái)審查程序的運(yùn)行情況以及各項(xiàng)指標(biāo)，及時(shí)做出錯(cuò)誤診斷并進(jìn)行相應(yīng)處理。內(nèi)置計(jì)數(shù)器（MapReduce相關(guān)、文件系統(tǒng)相關(guān)和作業(yè)調(diào)度相關(guān)） .也可以通過(guò) 查看Hello you, hello me的計(jì)數(shù)器信息 Counters: 19 File Output Format Counters Bytes

2、Written=19/reduce輸出到hdfs的字節(jié)數(shù) FileSystemCounters FILE_BYTES_READ=481 HDFS_BYTES_READ=38 FILE_BYTES_WRITTEN=81316 HDFS_BYTES_WRITTEN=19 File Input Format Counters Bytes Read=19/map從hdfs讀取的字節(jié)數(shù) Map-Reduce Framework Map output materialized bytes=49 Map input records=2 /map讀入的記錄行數(shù) Reduce shuffle bytes=0 S

3、pilled Records=8 Map output bytes=35 Total committed heap usage (bytes)=266469376 SPLIT_RAW_BYTES=105 Combine input records=0 Reduce input records=4/reduce從map端接收的記錄行數(shù) Reduce input groups=3/reduce函數(shù)接收的key數(shù)量，即歸并后的k2數(shù)量 Combine output records=0 Reduce output records=3/reduce輸出的記錄行數(shù) Map output records=4

4、/map輸出的記錄行數(shù)自定義計(jì)數(shù)器與實(shí)現(xiàn)Context類調(diào)用方法getCounter()org.apache.hadoop.mapreduce.Counter計(jì)數(shù)器聲明1.通過(guò)枚舉聲明context.getCounter(Enum enum) 2.動(dòng)態(tài)聲明context.getCounter(String groupName,String counterName) 計(jì)數(shù)器操作counter.setValue(long value);/設(shè)置初始值counter.increment(long incr);/增加計(jì)數(shù)Partitioner是partitioner的基類，如果需要定制partition

5、er也需要繼承該類。2. HashPartitioner是mapreduce的默認(rèn)partitioner。計(jì)算方法是which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到當(dāng)前的目的reducer。3. (例子以jar形式運(yùn)行) Partitioner編程在map和reduce階段進(jìn)行排序時(shí)，比較的是k2。v2是不參與排序比較的。如果要想讓v2也進(jìn)行排序，需要把k2和v2組裝成新的類，作為k2，才能參與比較。分組時(shí)也是按照k2進(jìn)行比較的。排序和分組Combiners編程 每一個(gè)map可能會(huì)產(chǎn)生大量的輸出，com

6、biner的作用就是在map端對(duì)輸出先做一次合并，以減少傳輸?shù)絩educer的數(shù)據(jù)量。 combiner最基本是實(shí)現(xiàn)本地key的歸并，combiner具有類似本地的reduce功能。 如果不用combiner，那么，所有的結(jié)果都是reduce完成，效率會(huì)相對(duì)低下。使用combiner，先完成的map會(huì)在本地聚合，提升速度。注意：Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計(jì)算結(jié)果。所以從我的想法來(lái)看，Combiner只應(yīng)該用于那種Reduce的輸入key/value與輸出key/value類型完全一致，且不影響最終結(jié)果的場(chǎng)景。比如累加，最大值等。Shuffle

7、每個(gè)map有一個(gè)環(huán)形內(nèi)存緩沖區(qū)，用于存儲(chǔ)任務(wù)的輸出。默認(rèn)大小100MB（io.sort.mb屬性），一旦達(dá)到閥值0.8（io.sort.spill.percent）,一個(gè)后臺(tái)線程把內(nèi)容寫到(spill)磁盤的指定目錄（mapred.local.dir）下的新建的一個(gè)溢出寫文件。寫磁盤前，要partition,sort。如果有combiner，combine排序后數(shù)據(jù)。等最后記錄寫完，合并全部溢出寫文件為一個(gè)分區(qū)且排序的文件。Reducer通過(guò)Http方式得到輸出文件的分區(qū)。TaskTracker為分區(qū)文件運(yùn)行Reduce任務(wù)。復(fù)制階段把Map輸出復(fù)制到Reducer的內(nèi)存或磁盤。一個(gè)Map任務(wù)

8、完成，Reduce就開(kāi)始復(fù)制輸出。排序階段合并map輸出。然后走Reduce階段。hadoop的壓縮codecCodec為壓縮，解壓縮的算法實(shí)現(xiàn)。 在Hadoop中，codec由CompressionCode的實(shí)現(xiàn)來(lái)表示。下面是一些實(shí)現(xiàn)：壓縮算法效率比較MapReduce的輸出進(jìn)行壓縮輸出的壓縮屬性：MapReduce中的join操作reduce side joinmap side joinreduce side joinreduce side join是一種最簡(jiǎn)單的join方式，其主要思想如下：在map階段，map函數(shù)同時(shí)讀取兩個(gè)文件File1和File2，為了區(qū)分兩種來(lái)源的key/valu

9、e數(shù)據(jù)對(duì)，對(duì)每條數(shù)據(jù)打一個(gè)標(biāo)簽（tag）,比如：tag=0表示來(lái)自文件File1，tag=2表示來(lái)自文件File2。即：map階段的主要任務(wù)是對(duì)不同文件中的數(shù)據(jù)打標(biāo)簽。在reduce階段，reduce函數(shù)獲取key相同的來(lái)自File1和File2文件的value list， 然后對(duì)于同一個(gè)key，對(duì)File1和File2中的數(shù)據(jù)進(jìn)行join（笛卡爾乘積）。即：reduce階段進(jìn)行實(shí)際的連接操作。map side join之所以存在reduce side join，是因?yàn)樵趍ap階段不能獲取所有需要的join字段，即：同一個(gè)key對(duì)應(yīng)的字段可能位于不同map中。Reduce side join是

10、非常低效的，因?yàn)閟huffle階段要進(jìn)行大量的數(shù)據(jù)傳輸。Map side join是針對(duì)以下場(chǎng)景進(jìn)行的優(yōu)化：兩個(gè)待連接表中，有一個(gè)表非常大，而另一個(gè)表非常小，以至于小表可以直接存放到內(nèi)存中。這樣，我們可以將小表復(fù)制多份，讓每個(gè)map task內(nèi)存中存在一份（比如存放到hash table中），然后只掃描大表：對(duì)于大表中的每一條記錄key/value，在hash table中查找是否有相同的key的記錄，如果有，則連接后輸出即可。為了支持文件的復(fù)制，Hadoop提供了一個(gè)類DistributedCache，使用該類的方法如下：（1）用戶使用靜態(tài)方法DistributedCache.addCach

11、eFile()指定要復(fù)制的文件，它的參數(shù)是文件的URI（如果是HDFS上的文件，可以這樣：hdfs:/namenode:9000/home/XXX/file，其中9000是自己配置的NameNode端口號(hào)）。JobTracker在作業(yè)啟動(dòng)之前會(huì)獲取這個(gè)URI列表，并將相應(yīng)的文件拷貝到各個(gè)TaskTracker的本地磁盤上。（2）用戶使用DistributedCache.getLocalCacheFiles()方法獲取文件目錄，并使用標(biāo)準(zhǔn)的文件讀寫API讀取相應(yīng)的文件。調(diào)度算法hadoop目前支持以下三種調(diào)度器：FifoScheduler：最簡(jiǎn)單的調(diào)度器，按照先進(jìn)先出的方式處理應(yīng)用。只有一個(gè)隊(duì)列

12、可提交應(yīng)用，所有用戶提交到這個(gè)隊(duì)列。沒(méi)有應(yīng)用優(yōu)先級(jí)可以配置。CapacityScheduler：可以看作是FifoScheduler的多隊(duì)列版本。每個(gè)隊(duì)列可以限制資源使用量。但是，隊(duì)列間的資源分配以使用量作排列依據(jù)，使得容量小的隊(duì)列有競(jìng)爭(zhēng)優(yōu)勢(shì)。集群整體吞吐較大。延遲調(diào)度機(jī)制使得應(yīng)用可以放棄跨機(jī)器或者跨機(jī)架的調(diào)度機(jī)會(huì)，爭(zhēng)取本地調(diào)度。 詳情見(jiàn)官網(wǎng) FairScheduler：多隊(duì)列，多用戶共享資源。特有的客戶端創(chuàng)建隊(duì)列的特性，使得權(quán)限控制不太完美。根據(jù)隊(duì)列設(shè)定的最小共享量或者權(quán)重等參數(shù)，按比例共享資源。延遲調(diào)度機(jī)制跟CapacityScheduler的目的類似，但是實(shí)現(xiàn)方式稍有不同。資源搶占特性，

13、是指調(diào)度器能夠依據(jù)公平資源共享算法，計(jì)算每個(gè)隊(duì)列應(yīng)得的資源，將超額資源的隊(duì)列的部分容器釋放掉的特性。 詳情見(jiàn)官網(wǎng) 配置FairScheduler修改mapred-site.xml，然后重啟集群更多配置見(jiàn)conf/fair-scheduler.xml配置CapacityScheduler修改mapred-site.xml，然后重啟集群更多配置見(jiàn)conf/capacity-scheduler.xmlCapacityScheduler運(yùn)行結(jié)果Hadoop StreamingHadoop Streaming是Hadoop提供的一個(gè)編程工具，它允許用戶使用任何可執(zhí)行文件或者腳本文件作為Mapper和Re

14、ducer，通過(guò)Hadoop Streaming編寫的MapReduce應(yīng)用程序中每個(gè)任務(wù)可以由不同的編程語(yǔ)言環(huán)境組成MapReduce的job如何調(diào)優(yōu)如果存在大量的小數(shù)據(jù)文件，可以使用SequenceFile、自定義的CombineFileInputFormat推測(cè)執(zhí)行在整個(gè)集群上關(guān)閉，特定需要的作業(yè)單獨(dú)開(kāi)啟，一般可以省下約5%10%的集群資源 mapred.map.task.speculative.execution=true; mapred.reduce.task.speculative.execution=false; 開(kāi)啟jvm重用 mapred.job.reuse.jvm.num.tasks=-1 增加InputSplit大小 mapred.min.split.size=268435456增大map輸出的緩存 io.sort.mb=300增加合并spill文件數(shù)量 io.sort.factor=50map端輸出壓縮，推薦LZO壓縮算法 press.map.output=true; pression.lzo.LzoCodec; 增大shuffle復(fù)制線程數(shù) mapred.reduce.parallel.