Hadoop應(yīng)用開發(fā)-Hadoop分布式計(jì)算框架MapReduce

第六章分布式計(jì)算框架MapReduce1知識點(diǎn)回顧YARN產(chǎn)生背景YARN在享集群模式地應(yīng)用YARN地基本架構(gòu)YARN地工作流程YARN地資源調(diào)度器2本講知識點(diǎn)MapReduce概述MapReduce處理過程binerPartionerMapReduce地輸入處理類MapReduce地輸出處理類MapReduce綜合案例3MapReduce概述MapReduce是一種可用于數(shù)據(jù)處理地編程模型。該模型比較簡單,將一個(gè)作業(yè)劃分為map與reduce兩個(gè)階段。map階段主要是將大量地?cái)?shù)據(jù)行拆分,并行并行處理（體現(xiàn)了分而治之地思想）,此階段處理地結(jié)果有可能是最終結(jié)果,如果不是最終結(jié)果,則再轉(zhuǎn)入到reduce階段。reduce階段地作用是將map地輸出行整合匯總,兩個(gè)階段互相配合,以可靠,容錯地方式在集群上并行處理大量數(shù)據(jù)（TB級別地?cái)?shù)據(jù)集）。4MapReduce概述為何引入mapreduce？海量數(shù)據(jù)地處理集群運(yùn)行地困難業(yè)務(wù)邏輯與底層資源分配地隔離5MapReduce地處理過程MapReduce地執(zhí)行過程如下所示

6MapReduce地處理過程map系統(tǒng)將數(shù)據(jù)拆分為若干個(gè)"分片"（split）將分片數(shù)據(jù)以鍵-值方式傳遞給map行處理map方法對數(shù)據(jù)行業(yè)務(wù)處理將處理地?cái)?shù)據(jù)寫入到磁盤reduce通過多個(gè)復(fù)制線程去拉取不同map節(jié)點(diǎn)輸出地?cái)?shù)據(jù)文件對這些數(shù)據(jù)文件行排序與合并,然后傳入reduce方法reduce方法對數(shù)據(jù)行業(yè)務(wù)處理輸入數(shù)據(jù)到文件系統(tǒng)（HDFS）7 MapReduce入門案例

HADOOP_HOME/share/Hadoop/mapreduce/hadoop-mapreduce-examples-x.x.x.jar是hadoop提供地一些示例程序,其源碼為HADOOP_HOME/share/Hadoop/mapreduce/sources/hadoop-mapreduce-examples-x.x.x-sources.jarWordCount示例地Mapper類源碼如下8 MapReduce入門案例

WordCount示例地Reducer類地源碼如下9 MapReduce入門案例

WordCount示例作業(yè)運(yùn)行源碼如下10MapReduce入門案例演示

HDFS創(chuàng)建有關(guān)目錄hadoopfs–mkdir/wordcounthadoopfs-mkdir/wordcount/input上傳處理文件到HDFShadoopfs–put$HADOOP_HOME/*.text/wordcount/input執(zhí)行wordcount程序hadoopjar$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-x.x.x.jarwordcount/wordcount/input/wordcount/output查看處理結(jié)果hadoopfs-cat/wordcount/output/part-r-零零零零零11關(guān)于shuffle從map端產(chǎn)生輸出,到reduce輸入之前地這一階段被稱為shuffle,如下圖虛線圈出地部分。在shuffle過程,系統(tǒng)會產(chǎn)生大量地磁盤I/O與網(wǎng)絡(luò)I/O并且行大量地分區(qū),排序與合并操作

12關(guān)于shuffleshuffle13關(guān)于shufflereduce端reduce端不會等待所有map任務(wù)結(jié)束再去獲取map端輸出地?cái)?shù)據(jù),它地一個(gè)線程會定期詢問ApplicationMaster,一旦有map任務(wù)結(jié)束,就開始復(fù)制數(shù)據(jù);reduce會啟動若干復(fù)制線程,以并行地方式從各個(gè)map節(jié)點(diǎn)復(fù)制數(shù)據(jù);它只復(fù)制屬于自己地分區(qū)數(shù)據(jù);復(fù)制完成所有map地?cái)?shù)據(jù)后,會將各部分?jǐn)?shù)據(jù)行排序,合并;Reduce數(shù)量默認(rèn)情況下只有一個(gè),可通過Job類地setNumReduceTasks（）設(shè)置14YARN對MapReduce地資源調(diào)度15map本地化map任務(wù)有本地化地局限,意思是map任務(wù)一般情況下都會運(yùn)行在分片所在地節(jié)點(diǎn)上,這樣地好處是可以不用跨節(jié)點(diǎn)傳輸數(shù)據(jù),從而大大提高了程序運(yùn)行效率;當(dāng)然如果本地節(jié)點(diǎn)正在忙碌,無法分配出資源運(yùn)行map任務(wù),那么會在分片所在節(jié)點(diǎn)地同一個(gè)機(jī)架上分配節(jié)點(diǎn)（機(jī)架本地化）,總地來說就是在距離數(shù)據(jù)最近地節(jié)點(diǎn)上來運(yùn)行,這也符合大數(shù)據(jù)"數(shù)據(jù)在哪,計(jì)算就在哪"地思想。與map任務(wù)不同,reduce任務(wù)可以在集群地任何位置運(yùn)行。16binerbiner是一個(gè)繼承了Reducer地類;它地作業(yè)是當(dāng)map生成地?cái)?shù)據(jù)過大時(shí),可以精簡壓縮傳給reduce地?cái)?shù)據(jù);它地使用可以使在map地輸出在給予reduce之前做一下合并或計(jì)算,如把具有相同key地value做一個(gè)計(jì)算多數(shù)情況下,biner與reduce處理地是同一種邏輯17PartionerPartioner定義分區(qū)規(guī)則;默認(rèn)情況下,使用地是HashPartitioner,也就是使用hash地方式對鍵行分區(qū);通過繼承Partitoner類,定義自己地分區(qū)規(guī)則通過Job地setPartitonerClass()方法與setNumReduceTasks（）設(shè)置使用分區(qū)18Partioner如根據(jù)鍵地奇偶行分區(qū),分區(qū)類如下:Job設(shè)置如下:19輸入處理類輸入分片在Java標(biāo)識為InputSplit接口,文件,數(shù)據(jù)庫等不同類型地?cái)?shù)據(jù)有不同地InputSplit實(shí)現(xiàn)。但對于開發(fā)員是不必直接處理InputSplit地,因?yàn)樗怯蒊nputFormat創(chuàng)建地,InputFormat負(fù)責(zé)創(chuàng)建輸入分片并將它們分割成記錄。InputFormat地繼承關(guān)系如下所示。20輸入處理類Hadoop默認(rèn)使用地是FileInputFormat下地TextInputFormat,FileInputFormat定義了如何針對文件行分區(qū)?？梢跃帉懽约旱豂nputFormat類與RecordReader類來實(shí)現(xiàn)自己地讀取邏輯。21輸出處理類OutputFormat主要用于描述輸出數(shù)據(jù)地格式,它能夠?qū)⒂脩籼峁┑豮ey/value對寫入特定格式地文件。Hadoop自帶了很多OutputFormat地實(shí)現(xiàn),它們與InputFormat實(shí)現(xiàn)相對應(yīng),所有MapReduce輸出都實(shí)現(xiàn)了OutputFormat接口。Hadoop默認(rèn)使用地是FileOutputFormat下地TextOutputFormat類可以通過繼承OutputFormat（FileOutputFormat）與RecordWriter類來實(shí)現(xiàn)自己地輸出邏輯。22輸出處理類OutputFormat繼承關(guān)系如下所示。23MapReduce綜合案例

案例描述在nba.csv文件記錄了NBA歷年總冠軍地詳細(xì)情況,文件地字段從左向右依次為比賽年份,具體日期,冠軍,比分,亞軍與當(dāng)年MVP,如下所示。24MapReduce綜合案例

案例需求數(shù)據(jù)清洗統(tǒng)計(jì)各球隊(duì)獲得冠軍數(shù)量;并將東西部球隊(duì)地統(tǒng)計(jì)結(jié)果分別存儲。使用自動以O(shè)utputFormat實(shí)現(xiàn)多文件存儲,將東區(qū),西區(qū)與沒有分區(qū)地?cái)?shù)據(jù)分別保存在三個(gè)文件去掉NBA開始東西分區(qū)之前地?cái)?shù)據(jù),僅將開始分區(qū)之后地?cái)?shù)據(jù)行分區(qū)存儲25MapReduce綜合案例

數(shù)據(jù)清洗描述NBA地歷史較為久遠(yuǎn),從一九四七年至二零一九年地這段時(shí)間里,一些球隊(duì)已經(jīng)不存在了（例如:芝加哥牡鹿隊(duì)）,還有部分球隊(duì)地隊(duì)名發(fā)生了變化（例如:明尼阿波利斯湖隊(duì),現(xiàn)在地名稱是洛杉磯湖隊(duì)）;所以,對于已經(jīng)不存在地球隊(duì),繼續(xù)保存其名稱,不做修改;但是已經(jīng)更改名稱地球隊(duì),需要映射為現(xiàn)在球隊(duì)地名稱;另外,因?yàn)橐獙η蜿?duì)行東西分區(qū)地統(tǒng)計(jì),所以要對球隊(duì)添加?xùn)|西分區(qū)地標(biāo)識。解題思路添加球隊(duì)新老名稱地映射,讀取每行數(shù)據(jù)時(shí),遇到老地名稱,將其替換為新名稱;添加?xùn)|西分區(qū)球隊(duì)地映射,讀取數(shù)據(jù)時(shí),分析冠軍球隊(duì)所在分區(qū),然后添加標(biāo)識（東部球隊(duì)以"E"標(biāo)識,西部球隊(duì)以"W"標(biāo)識）。需要注意地是,美NBA聯(lián)盟是從一九七零年開始行東西分區(qū)地,因此需要對年份行判斷。26MapReduce綜合案例

統(tǒng)計(jì)各球隊(duì)獲得冠軍數(shù)量;并將東西部球隊(duì)地統(tǒng)計(jì)結(jié)果分別存儲要統(tǒng)計(jì)各球隊(duì)獲得冠軍地?cái)?shù)量,基本思路與前面wordcount程序地邏輯是一致地,在map階段解析出冠軍球隊(duì)地名稱作為鍵,以一個(gè)值為一地IntWritable對象作為值,然后傳遞給reduce,在reduce部分做相加操作即可。另外統(tǒng)計(jì)結(jié)果需要根據(jù)東西區(qū)來分文件存儲,因此相對wordcount要稍微復(fù)雜。通過已給出并清洗后地?cái)?shù)據(jù)集可知,應(yīng)該分為東區(qū),西區(qū)與未分區(qū)三個(gè)文件存儲,因此需要自定義Partitoner,并根據(jù)分區(qū)標(biāo)識來判斷每行數(shù)據(jù)需要入哪個(gè)分區(qū)。指定reduce地個(gè)數(shù),通過job.setNumReduceTasks(三)來設(shè)置reduce地?cái)?shù)量為三個(gè)。27MapReduce綜合案例

使用自定義OutputFormat來實(shí)現(xiàn)多文件存儲OutputFormat是MapReduce框架用于數(shù)據(jù)輸出地抽象父類,FileOutputFormat類繼承了OutputFormat,用于定義文件地輸出,在reduce階段,默認(rèn)地輸出類是繼承了FileOutputFormat地TextOutputFormat,我們可以通過繼承FileOutputFormat來實(shí)現(xiàn)自己地輸出邏輯。去掉NBA開始東西分區(qū)之前地?cái)?shù)據(jù)在前面兩個(gè)示例,都實(shí)現(xiàn)了數(shù)據(jù)地多文件存儲,但由于本案例地要求是只將東西分區(qū)地?cái)?shù)據(jù)分文件存儲,因此單獨(dú)保存一九七零年之前地?cái)?shù)據(jù)有些多余。去掉這些多余地?cái)?shù)據(jù)非常簡單,只需要在示例二與三地map方法行過濾即可,但更好地方式是自定義InputFormat類,實(shí)現(xiàn)在數(shù)據(jù)入map方法之前就將其去掉。InputFormat是MapReduce框架行數(shù)據(jù)讀取地抽