Hadoop大數(shù)據(jù)原理與應(yīng)用實(shí)驗(yàn)教程

實(shí)驗(yàn)3準(zhǔn)備：分布式計(jì)算框架MapReduce實(shí)驗(yàn)3知識(shí)地圖實(shí)驗(yàn)3MapReduce編程一、實(shí)驗(yàn)?zāi)康?.理解MapReduce編程思想。2.理解MapReduce作業(yè)執(zhí)行流程。3.理解MR-App編寫(xiě)步驟，掌握使用MapReduceJavaAPI進(jìn)行MapReduce基本編程，熟練掌握如何在Hadoop集群上運(yùn)行MR-App并查看運(yùn)行結(jié)果。4.熟練掌握MapReduceWeb界面的使用。5.掌握MapReduceShell常用命令的使用。二、實(shí)驗(yàn)環(huán)境本實(shí)驗(yàn)所需的軟件環(huán)境包括全分布模式Hadoop集群、Eclipse。三、實(shí)驗(yàn)內(nèi)容1.啟動(dòng)全分布模式Hadoop集群，守護(hù)進(jìn)程包括NameNode、DataNode、SecondaryNameNode、ResourceManager、NodeManager和JobHistoryServer。2.在Hadoop集群主節(jié)點(diǎn)上搭建MapReduce開(kāi)發(fā)環(huán)境Eclipse。3.查看Hadoop自帶的MR-App單詞計(jì)數(shù)源代碼WordCount.java，在Eclipse項(xiàng)目MapReduceExample下建立新包c(diǎn)om.xijing.mapreduce，模仿內(nèi)置的WordCount示例，自己編寫(xiě)一個(gè)WordCount程序，最后打包成JAR形式并在Hadoop集群上運(yùn)行該MR-App，查看運(yùn)行結(jié)果。4分別在自編MapReduce程序WordCount運(yùn)行過(guò)程中和運(yùn)行結(jié)束后查看MapReduceWeb界面。5.分別在自編MapReduce程序WordCount運(yùn)行過(guò)程中和運(yùn)行結(jié)束后練習(xí)MapReduceShell常用命令。6.關(guān)閉Hadoop集群。實(shí)驗(yàn)3準(zhǔn)備：分布式計(jì)算框架MapReduce3.1MapReduce編程思想3.2MapReduce體系架構(gòu)3.3MapReduce作業(yè)執(zhí)行流程3.4MapReduce數(shù)據(jù)類型與格式3.5MapReduce接口3.6其它主流分布式計(jì)算框架3.1MapReduce編程思想移動(dòng)計(jì)算比移動(dòng)數(shù)據(jù)更劃算Google論文分而治之3.1MapReduce編程思想MapReduce是Hadoop生態(tài)中的一款分布式計(jì)算框架，它可以讓不熟悉分布式計(jì)算的人員也能編寫(xiě)出優(yōu)秀的分布式系統(tǒng)，因此可以讓開(kāi)發(fā)人員將精力專注到業(yè)務(wù)邏輯本身。MapReduce采用“分而治之”的核心思想，可以先將一個(gè)大型任務(wù)拆分成若干個(gè)簡(jiǎn)單的子任務(wù)，然后將每個(gè)子任務(wù)交給一個(gè)獨(dú)立的節(jié)點(diǎn)去處理。當(dāng)所有節(jié)點(diǎn)的子任務(wù)都處理完畢后，再匯總所有子任務(wù)的處理結(jié)果，從而形成最終的結(jié)果。以“單詞統(tǒng)計(jì)”為例，如果要統(tǒng)計(jì)一個(gè)擁有海量單詞的詞庫(kù)，就可以先將整個(gè)詞庫(kù)拆分成若干個(gè)小詞庫(kù)，然后將各個(gè)小詞庫(kù)發(fā)送給不同的節(jié)點(diǎn)去計(jì)算，當(dāng)所有節(jié)點(diǎn)將分配給自己的小詞庫(kù)中的單詞統(tǒng)計(jì)完畢后，再將各個(gè)節(jié)點(diǎn)的統(tǒng)計(jì)結(jié)果進(jìn)行匯總，形成最終的統(tǒng)計(jì)結(jié)果。以上，“拆分”任務(wù)的過(guò)程稱為Map階段，“匯總”任務(wù)的過(guò)程稱為Reduce階段。3.1MapReduce編程思想節(jié)點(diǎn)3海量詞庫(kù)小詞庫(kù)小詞庫(kù)小詞庫(kù)統(tǒng)計(jì)部分單詞統(tǒng)計(jì)全部單詞Map階段Reduce階段節(jié)點(diǎn)1節(jié)點(diǎn)2統(tǒng)計(jì)部分單詞節(jié)點(diǎn)4節(jié)點(diǎn)5MapReduce執(zhí)行流程3.2MapReduce體系架構(gòu)舊版MapReduce（MapReduce1.0）采用的是典型的Master/Slave結(jié)構(gòu)，Master表現(xiàn)為JobTracker進(jìn)程，而Slave表現(xiàn)為TaskTracker。

JobTrackerTaskTrackerClientClientTaskSchedulerMapTaskMapTaskReduceTask

TaskTrackerMapTaskMapTaskReduceTask

TaskTrackerMapTaskMapTaskReduceTaskMapReduce1.0體系架構(gòu)3.2MapReduce體系架構(gòu)MapReduce1.0架構(gòu)過(guò)于簡(jiǎn)單，例如Master的任務(wù)過(guò)于集中，并且存在單點(diǎn)故障等問(wèn)題。因此，MapReduce進(jìn)行了一次重要的升級(jí)，舍棄JobTracker和TaskTracker，而改用了ResourceManager進(jìn)程負(fù)責(zé)處理資源，并且使用ApplicationMaster進(jìn)程管理各個(gè)具體的應(yīng)用，用NodeManager進(jìn)程對(duì)各個(gè)節(jié)點(diǎn)的工作情況進(jìn)行監(jiān)聽(tīng)。升級(jí)后的MapReduce稱為MapReduce2.0。3.2MapReduce體系架構(gòu)ResourceManagerNameNodeNodeManagerApplicationMasterDataNodeNodeManagerApplicationMasterDataNodeNodeManagerContainerDataNodeContainerNodeManagerContainerDataNodeNodeManagerContainerDataNodeNodeManagerContainerDataNodeClientClientMapReduce2.0執(zhí)行作業(yè)時(shí)體系架構(gòu)3.3MapReduce作業(yè)執(zhí)行流程最終結(jié)果<key,value>中間結(jié)果<key,List(value)>中間結(jié)果<key,value>輸入<key,value>加載文件最終結(jié)果<key,value>中間結(jié)果<key,List(value)>中間結(jié)果<key,value>輸入<key,value>寫(xiě)入文件分布式文件系統(tǒng)（如HDFS）InputFormatSplitSplitSplitRRRRRRMapMapMapShuffleReduceOutputFormat節(jié)點(diǎn)1加載文件寫(xiě)入文件InputFormatSplitSplitSplitRRRRRRMapMapMapShuffleReduceOutputFormat節(jié)點(diǎn)2分布式文件系統(tǒng)（如HDFS）3.3MapReduce作業(yè)執(zhí)行流程（1）InputFormatInputFormat模塊首先對(duì)輸入數(shù)據(jù)做預(yù)處理，比如驗(yàn)證輸入格式是否符合輸入定義；然后將輸入文件切分為邏輯上的多個(gè)InputSplit，InputSplit是MapReduce對(duì)文件進(jìn)行處理和運(yùn)算的輸入單位，并沒(méi)有對(duì)文件進(jìn)行實(shí)際切割；由于InputSplit是邏輯切分而非物理切分，所以還需要通過(guò)RecordReader（圖4-4中的RR）根據(jù)InputSplit中的信息來(lái)處理InputSplit中的具體記錄，加載數(shù)據(jù)并轉(zhuǎn)換為適合Map任務(wù)讀取的鍵值對(duì)<key,valule>，輸入給Map任務(wù)。（2）MapMap模塊會(huì)根據(jù)用戶自定義的映射規(guī)則，輸出一系列的<key,value>作為中間結(jié)果。（3）Shuffle為了讓Reduce可以并行處理Map的結(jié)果，需要對(duì)Map的輸出進(jìn)行一定的排序、分區(qū)、合并、歸并等操作，得到<key,List(value)>形式的中間結(jié)果，再交給對(duì)應(yīng)的Reduce進(jìn)行處理，這個(gè)過(guò)程叫做Shuffle。（4）ReduceReduce以一系列的<key,List(value)>中間結(jié)果作為輸入，執(zhí)行用戶定義的邏輯，輸出<key,valule>形式的結(jié)果給OutputFormat。（5）OutputFormatOutputFormat模塊會(huì)驗(yàn)證輸出目錄是否已經(jīng)存在以及輸出結(jié)果類型是否符合配置文件中的配置類型，如果都滿足，就輸出Reduce的結(jié)果到分布式文件系統(tǒng)。一個(gè)WordCount執(zhí)行過(guò)程的實(shí)例HDFS文件file.txtHelloWorldHelloHadoopHelloMapReduceInput<1,”HelloWorld”><2,”HelloHadoop”><3,”HelloMapReduce”>split<”Hello”，1><“World”,1><“Hello”,1><“Hadoop”,1><“Hello”,1><”MapReduce”,1>Map<“Hadoop”,1><”Hello”，<1,1,1>><”MapReduce”,1><“World”,1>Shuffle<“Hadoop”,1><”Hello”，3><”MapReduce”,1><“World”,1>ReduceHDFS文件（part-r-00000）Output3.4MapReduce數(shù)據(jù)類型與格式MapReduce使用了Text定義字符串，使用了IntWritable定義整型變量，而沒(méi)有使用Java內(nèi)置的String和int類型。這樣做主要是有兩個(gè)方面的原因：MapReduce是集群運(yùn)算，因此必然會(huì)在執(zhí)行期間進(jìn)行網(wǎng)絡(luò)傳輸，然而在網(wǎng)絡(luò)中傳輸?shù)臄?shù)據(jù)必須是可序列化的類型。為了良好地匹配MapReduce內(nèi)部的運(yùn)行機(jī)制，MapReduce就專門設(shè)計(jì)了一套數(shù)據(jù)類型。MapReduce中常見(jiàn)數(shù)據(jù)類型數(shù)據(jù)類型說(shuō)明IntWritable整型類型LongWritable長(zhǎng)整型類型FloatWritable單精度浮點(diǎn)數(shù)類型DoubleWritable雙精度浮點(diǎn)數(shù)類型ByteWritable字節(jié)類型BooleanWritable布爾類型TextUTF-8格式存儲(chǔ)的文本類型NullWritable空對(duì)象3.4MapReduce數(shù)據(jù)類型與格式需要注意的是，這些數(shù)據(jù)類型的定義類都實(shí)現(xiàn)了WritableComparable接口，其源碼如下所示。publicabstractinterfaceWritableComparableextendsWritable,Comparable{...}可以發(fā)現(xiàn)，WritableComparable繼承自Writable和Comparable接口。其中Writable就是MapReduce提供的序列化接口（類似于Java中的Serializable接口），源碼如下所示。publicabstractinterfaceorg.apache.hadoop.io.Writable{publicabstractvoidwrite(DataOutputoutput)throwsIOException;publicabstractvoidreadFields(DataInputinput)throwsjava.io.IOException;...}其中，write()用于將數(shù)據(jù)進(jìn)行序列化操作；readFields()用于將數(shù)據(jù)進(jìn)行反序列化操作。Comparable接口就是Java中的比較器，用于對(duì)數(shù)據(jù)集進(jìn)行排序操作。因此，如果我們要在MapReduce自定義一個(gè)數(shù)據(jù)類型，就需要實(shí)現(xiàn)Writable接口；如果還需要對(duì)自定義的數(shù)據(jù)類型進(jìn)行排序操作，就需要實(shí)現(xiàn)WritableComparable接口（或者分別實(shí)現(xiàn)Writable和Comparable接口）。3.5MapReduce接口1.MapReduceWebUI2.MapReduceShell3.MapReduceJavaAPI1.MapReduceWebUIMapReduceWebUI接口面向管理員。可以在頁(yè)面上看到已經(jīng)完成的所有MR-App執(zhí)行過(guò)程中的統(tǒng)計(jì)信息，該頁(yè)面只支持讀，不支持寫(xiě)。MapReduceWebUI的默認(rèn)地址為http://JobHistoryServerIP:19888，可以查看MapReduce的歷史運(yùn)行情況。1.MapReduceWebUIMapReduce歷史情況2.MapReduceShellMapReduceShell接口面向MapReduce程序員。程序員通過(guò)Shell接口能夠向YARN集群提交MR-App，查看正在運(yùn)行的MR-App，甚至可以終止正在運(yùn)行的MR-App。MapReduceShell命令統(tǒng)一入口為：mapred，語(yǔ)法格式如下：mapred[--configconfdir][--loglevelloglevel]COMMAND讀者需要注意的是，若$HADOOP_HOME/bin未加入到系統(tǒng)環(huán)境變量PATH中，則需要切換到Hadoop安裝目錄下，輸入“bin/mapred”?！癿apred”幫助2.MapReduceShell1）用戶命令2）管理員命令1）用戶命令命令選項(xiàng)功能描述archive創(chuàng)建一個(gè)Hadoop檔案文件archive-logs將聚合日志合并到Hadoop檔案文件中classpath打印運(yùn)行MapReduce子命令所需的包路徑distcp遞歸拷貝文件或目錄job管理MapReduce作業(yè)pipes運(yùn)行Pipes任務(wù)，此功能允許用戶使用C++語(yǔ)言編寫(xiě)MapReduce程序queue查看JobQueue信息2）管理員命令命令選項(xiàng)功能描述historyserver啟動(dòng)JobHistoryServer服務(wù)hsadminJobHistoryServer管理命令接口其中，命令“mapredhistoryserver”與啟動(dòng)MapReduce的命令“mr-jobhistory-daemon.shstarthistoryserver”效果相同。關(guān)于MapReduceShell命令的完整說(shuō)明，讀者請(qǐng)參考官方網(wǎng)站/docs/r2.9.2/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapredCommands.html。3.MapReduceJavaAPIMapReduceJavaAPI接口面向Java開(kāi)發(fā)工程師。程序員可以通過(guò)該接口編寫(xiě)MR-App用戶層代碼MRApplicationBusinessLogic?；赮ARN編寫(xiě)的MR-App和基于MapReduce1.0編寫(xiě)的MR-App編程步驟相同。MR-App稱為MapReduce應(yīng)用程序，標(biāo)準(zhǔn)YARN-App包含3部分：MRv2框架中的MRAppMaster、MRClient，加上用戶編寫(xiě)的MRApplicationBusinessLogic（Mapper類和Reduce類），合稱為MR-App。3.MapReduceJavaAPIMR-App編寫(xiě)步驟如下所示：（1）編寫(xiě)MRApplicationBusinessLogic。自行編寫(xiě)。（2）編寫(xiě)MRApplicationMaster。無(wú)需編寫(xiě)，Hadoop開(kāi)發(fā)人員已編寫(xiě)好MRAppMaster.java。（3）編寫(xiě)MRApplicationClient。無(wú)需編寫(xiě)，Hadoop開(kāi)發(fā)人員已編寫(xiě)好YARNRunner.java。其中，MRApplicationBusinessLogic編寫(xiě)步驟如下：（1）確定<key,value>對(duì)。（2）定制輸入格式。（3）Mapper階段，其業(yè)務(wù)代碼需要繼承自org.apache.hadoop.mapreduce.Mapper類。（4）Reducer階段，其業(yè)務(wù)代碼需要繼承自org.apache.hadoop.mapreduce.Reducer類。（5）定制輸出格式。3.MapReduceJavaAPI編寫(xiě)類后，在main方法里，按下述過(guò)程依次指向各類即可：（1）實(shí)例化配置文件類。（2）實(shí)例化Job類。（3）指向InputFormat類。（4）指向Mapper類。（5）指向Partitioner類。（6）指向Reducer類。（7）指向OutputFormat類。（8）提交任務(wù)。MapReduceJavaAPI常用類類名說(shuō)明org.apache.hadoop.mapreduce.JobMapReduce作業(yè)類org.apache.hadoop.mapreduce.MapperMapper類，泛型類，帶有4個(gè)參數(shù)，分別表示Map階段輸入數(shù)據(jù)的key類型、輸入數(shù)據(jù)的value類型、輸出數(shù)據(jù)的key類型、輸出數(shù)據(jù)的value類型。其中，輸入的key為Object(默認(rèn)是行)，輸入的值為Text（Hadoop中的String類型），輸出的key為Text（關(guān)鍵字），輸出的值為IntWritable（Hadoop中的int類型）org.apache.hadoop.mapreduce.ReducerReducer類，泛型類，帶有4個(gè)參數(shù)，分別表示Reduce階段輸入數(shù)據(jù)的key類型、value類型，輸出數(shù)據(jù)的key類型、value類型org.apache.hadoop.mapreduce.InputFormatMapReduce接收輸入數(shù)據(jù)的頂級(jí)類org.apache.hadoop.mapreduce.OutputFormatMapReduce接收輸出數(shù)據(jù)的頂級(jí)類關(guān)于MapReduceAPI的完整說(shuō)明，讀者請(qǐng)參考官方網(wǎng)站/docs/r2.9.2/api/index.html。MapReduceJavaAPI3.6其它主流分布式計(jì)算框架1.SparkStreamingSparkStreaming在時(shí)效性方面做了改進(jìn)，它使得計(jì)算的中間結(jié)果可以保存在內(nèi)存中，從而大大提高了計(jì)算速度，可以用于開(kāi)發(fā)延遲性較低的系統(tǒng)例如推薦系統(tǒng)等。SparkStreaming是ApacheSpark的內(nèi)容之一，ApacheSpark是一款快速處理大數(shù)據(jù)的計(jì)算引擎。Spark生態(tài)系統(tǒng)SparkSQLSparkStreamingMLlib(machinelearning)GraphX(graph)ApacheSpark

HDFSTachyonYARNMesos應(yīng)用接口層計(jì)算引擎層集群資源管理層存儲(chǔ)層StandaloneKubernetes1.SparkStreamingSparkCore是Spark的核心基礎(chǔ)，包含了彈性分布式數(shù)據(jù)集（即RDD）等核心組件。但需要注意的是，SparkCore是離線計(jì)算的，這點(diǎn)就類似于MapReduce的處理過(guò)程。而SparkStreaming則是將連續(xù)的數(shù)據(jù)轉(zhuǎn)換為不連續(xù)的離散流（DStream），從而實(shí)現(xiàn)了快速的數(shù)據(jù)處理功能。SparkSQL則用于簡(jiǎn)化Spark庫(kù)，就好比可以使用Hive簡(jiǎn)化MapReduce一樣，我們可以使用SparkSQL快速實(shí)現(xiàn)Spark開(kāi)發(fā)，具體地講，SparkSQL可以將DStream轉(zhuǎn)為Spark處理時(shí)的RDD，然后運(yùn)行RDD程序。Spark生態(tài)中的其他技術(shù)也都各自擅長(zhǎng)的領(lǐng)域。例如MLlib是Spark提供的機(jī)器學(xué)習(xí)類庫(kù)，而GraphX則是一款圖計(jì)算組件，可以實(shí)現(xiàn)對(duì)圖形數(shù)據(jù)的分析、圖形數(shù)據(jù)可視化等功能。在實(shí)際開(kāi)發(fā)時(shí)，Spark除了使用Java開(kāi)發(fā)以外，還可以使用Scala、Python等語(yǔ)言開(kāi)發(fā)。2.StormTwitter開(kāi)源的ApacheStorm在實(shí)時(shí)處理領(lǐng)域也有著廣泛的應(yīng)用，它可以用于分布式或大數(shù)據(jù)領(lǐng)域的實(shí)時(shí)計(jì)算，其Logo如圖4-24所示。Storm在運(yùn)行時(shí)，是由一個(gè)Nimbus和一個(gè)或多個(gè)supervisors組成，并且通過(guò)ApacheZooKeeper進(jìn)行各組件之間的協(xié)調(diào)。在運(yùn)行方式上，Storm也有本地模式和生成模式兩種。一般情況下，在平時(shí)的開(kāi)發(fā)、調(diào)試過(guò)程中，可以使用本地模式，這樣利于快速開(kāi)發(fā)，以及進(jìn)行性能的優(yōu)化。而在最終實(shí)施時(shí)，就可以切換到生產(chǎn)模式，最大限度地發(fā)揮Storm的作用和提高Storm的性能。3.FlinkFlink和Storm的定位就是分布式實(shí)時(shí)計(jì)算框架，因此二者的時(shí)效性最高，可以用于實(shí)時(shí)分析、實(shí)時(shí)計(jì)算等領(lǐng)域。具體地講，可以將Flink稱為一個(gè)流式的數(shù)據(jù)執(zhí)行引擎，可以提供高性能的分布式數(shù)據(jù)通信以及容錯(cuò)機(jī)制。并