《大數據技術實戰案例教程》實驗指導書實驗3MapReduce編程編寫者：徐魯輝實驗3MapReduce編程3.1實驗目的1.理解MapReduce編程思想。2.理解MapReduce作業執行流程。3.理解MR-App編寫步驟，掌握使用MapReduceJavaAPI進行MapReduce基本編程，熟練掌握如何在Hadoop集群上運行MR-App并查看運行結果。4.熟練掌握MapReduceWebUI界面的使用。5.掌握MapReduceShell常用命令的使用。3.2實驗環境本實驗所需的軟件環境包括全分布模式Hadoop集群、Eclipse。3.3實驗內容1.啟動全分布模式Hadoop集群，守護進程包括NameNode、DataNode、SecondaryNameNode、ResourceManager、NodeManager和JobHistoryServer。2.在Eclipse下創建Java項目MapReduceExample，在其下建立新包com.xijing.mapreduce，編寫MapReduce程序，已知某個超市的結算記錄，從左往右各字段的含義依次是會員編號、結算時間、消費金額和用戶身份，要求計算出會員和非會員的平均消費金額。最后打包成JAR形式并在Hadoop集群上運行該MR-App，查看運行結果。3.分別在自編MapReduce程序運行過程中和運行結束后練習MapReduceShell常用命令。4.分別在自編MapReduce程序運行過程中和運行結束后查看MapReduceWebUI界面。5.關閉Hadoop集群。3.4實驗原理3.4.1初識MapReduceMapReduce是一個可用于大規模數據處理分布式計算框架，HadoopMapReduce來源于Google公司2004年發表的一篇關于MapReduce的論文（MapReduce:SimplifiedDataProcessingonLargeClusters,2004），其對大數據并行處理采用“分而治之”的設計思想，并把函數式編程思想構建成抽象模型Map和Reduce。MapReduce在發展史上經過一次重大改變。舊版MapReduce（MapReduce1.0）采用典型的主從架構，主進程為JobTracker，從進程為TaskTracker，但是這種架構過于簡單，導致JobTracker的任務過于集中，并且存在單點故障等問題。因此，MapReduce進行了一次重要升級，舍棄了JobTracker和TaskTracker，而改用了ResourceManager進程負責處理資源，并且使用ApplicationMaster進程管理各個具體的應用，用NodeManager進程對各個節點的工作情況進行監聽，形成了純粹的計算框架MapReduce2.0和存粹的資源管理框架YARN。3.4.2MapReduce作業執行流程MapReduce作業的執行流程主要包括InputFormat、Map、Shuffle、Reduce、OutputFormat五個階段。InputFormat加載數據并轉換為適合Map任務讀取的鍵值對<key,valule>，輸入給Map任務；Map階段會根據用戶自定義的映射規則，輸出一系列的<key,value>作為中間結果；Shuffle階段對Map的輸出進行一定的排序、分區、合并、歸并等操作，得到<key,List(value)>形式的中間結果；Reduce階段以Shuffle輸出作為輸入，執行用戶定義的邏輯，輸出<key,valule>形式的結果給OutputFormat；OutputFormat模塊完成輸出驗證后將Reduce結果寫入分布式文件系統。MapReduce作業執行流程如圖3-2所示。最終結果最終結果<key,value>中間結果<key,List(value)>中間結果<key,value>輸入<key,value>加載文件最終結果<key,value>中間結果<key,List(value)>中間結果<key,value>輸入<key,value>寫入文件分布式文件系統（如HDFS）InputFormatSplitSplitSplitRRRRRRMapMapMapShuffleReduceOutputFormat節點1加載文件寫入文件InputFormatSplitSplitSplitRRRRRRMapMapMapShuffleReduceOutputFormat節點2分布式文件系統（如HDFS）圖3-2MapReduce作業執行流程3.4.3MapReduce入門案例WordCountHadoop提供了一個MapReduce入門案例“WordCount”，用于統計輸入文件中每個單詞出現的次數，其是最簡單也是最能體現MapReduce思想的程序之一，被稱為MapReduce版的“HelloWorld”。本節主要從源碼和執行流程兩個角度對MapReduce編程模型進行詳細分析。該案例源碼保存在$HADOOP_HOME/share/hadoop/mapreduce/sources/hadoop-mapreduce-examples-2.9.2.jar的WordCount.java中，其源碼共分為TokenizerMapper類、IntSumReducer類和main()函數三個部分。1.TokenizerMapper類從類名可知，該類是Map階段的實現。并且能夠發現，在MapReduce中Map階段的業務代碼需要繼承自org.apache.hadoop.mapreduce.Mapper類。Mapper類的四個泛型分別表示輸入數據的key類型、value類型，輸出數據的key類型、value類型。以本次“WordCount”為例，每次Map階段需要處理的數據是文件中的一行數據，而默認情況下這一行數據的偏移量（通俗的說，下一行記錄開始位置=上一行記錄的開始位置+上一行字符串內容的長度，這個相對字節的變化就叫做字節偏移量）就是輸入數據的key類型（一般而言，偏移量是一個長整型，也可以寫成本例中使用的Object類型）；輸入數據的value類型就是這行數據本身，因此是Text類型（即Hadoop中定義的字符串類型）；輸出數據的key類型是每個單詞本身，因此也是Text類型；而輸出數據的value類型，就表示該單詞出現了一次，因此就是數字1，可以表示為IntWritable類型（即Hadoop中定義的整數類型）。TokenizerMapper類的源碼如下所示。importorg.apache.hadoop.mapreduce.Mapper;publicstaticclassTokenizerMapperextendsMapper<Object,Text,Text,IntWritable>{privatefinalstaticIntWritableone=newIntWritable(1);privateTextword=newText();publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{StringTokenizeritr=newStringTokenizer(value.toString());while(itr.hasMoreTokens()){word.set(itr.nextToken());context.write(word,one);}}}從源碼可知，Mapper類提供了map()方法，用于編寫Map階段具體的業務邏輯。map()方法的前兩個參數表示輸入數據的key類型和value類型（即與Mapper類前兩個參數的含義一致），而map()的第三個參數Context對象表示Map階段的上下文對象，可以用于將Map階段的產物輸出到下一個階段中。2.IntSumReducer類數據在經過了Shuffle階段后，就會進入Reduce階段。入門案例“WordCount”中的Reduce源碼如下所示。importorg.apache.hadoop.mapreduce.Reducer;publicstaticclassIntSumReducerextendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritableval:values){sum+=val.get();}result.set(sum);context.write(key,result);}}從上述源碼可知，Reduce階段的業務代碼需要繼承自org.apache.hadoop.mapreduce.Reducer類。Reducer類的四個泛型分別表示Reduce階段輸入數據的key類型、value類型，以及Reduce階段輸出數據的key類型、value類型。MapReduce的流程依次Map階段、Shuffle階段和Reduce階段，因此Shuffle階段的產物就是Reduce階段的輸入數據。也就是說，本例中的IntSumReducer就是對Shuffle的產物進行了統計，即計算出了Flink、Hadoop、HDFS、Hello、Spark、University、Xijing各個單詞出現的次數分別是1、1、1、6、1、2、2。IntSumReducer源碼中最后的context.write()方法表示將Reduce階段的產物輸出到最終的HDFS中進行存儲，而存儲在HDFS中的具體位置是在main()方法中進行設置的。3.main()函數入門案例“WordCount”中main()函數的源碼如下所示。publicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();if(otherArgs.length<2){System.err.println("Usage:wordcount<in>[<in>...]<out>");System.exit(2);}Jobjob=Job.getInstance(conf,"wordcount");job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);for(inti=0;i<otherArgs.length-1;++i){FileInputFormat.addInputPath(job,newPath(otherArgs[i]));}FileOutputFormat.setOutputPath(job,newPath(otherArgs[otherArgs.length-1]));System.exit(job.waitForCompletion(true)?0:1);}從上述源碼可知，main()的輸入參數決定了輸入數據的文件位置，以及輸出數據存儲到HDFS中的位置。并且main()方法設置了Map階段和Reduce階段的類文件，以及通過setOutputKeyClass()和setOutputValueClass()指定了最終輸出數據的類型。4.向Hadoop集群提交并運行WordCount使用如下命令向Hadoop集群提交并運行WordCount，執行該命令前要求啟動Hadoop集群。[xuluhui@master~]$hadoopjar/usr/local/hadoop-2.9.2/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.9.2.jarwordcount/InputData/OutputData上述命令中，/InputData表示輸入目錄，/OutputData表示輸出目錄。執行該命令前，要求HDFS輸入目錄/InputData已存在，且HDFS輸出目錄/OutputData不存在，否則會拋出異常。部分執行過程如圖3-1所示。圖3-1向Hadoop集群提交并運行WordCount的執行過程（部分）上述程序執行完畢后，會將結果輸出到HDFS目錄/OutputData中，使用命令“hadoopfs-ls/OutputData”來查看，使用的HDFSShell命令及具體過程如圖3-2所示。圖3-2中/OutputData目錄下有2個文件，其中/OutputData/_SUCCESS表示Hadoop程序已執行成功，這個文件大小為0，文件名就告知了Hadoop程序的執行狀態；第二個文件/OutputData/part-r-00000才是Hadoop程序的運行結果。在命令終端利用命令“hadoopfs-cat/OutputData/part-r-00000”查看Hadoop程序的運行結果，使用的HDFSShell命令及單詞計數結果如圖3-2所示。圖3-2查看WordCount運行結果3.4.4MapReduce數據類型MapReduce是集群運算，因此要求在網絡中傳輸的數據必須是可序列化的類型，且為了良好匹配MapReduce內部運行機制，MapReduce專門設計了一套數據類型。例如MapReduce使用IntWritable定義整型變量，而不是Java內置的int類型。MapReduce中常見的數據類型如表3-1所示。表3-1MapReduce中常見數據類型數據類型說明IntWritable整型類型LongWritable長整型類型FloatWritable單精度浮點數類型DoubleWritable雙精度浮點數類型ByteWritable字節類型BooleanWritable布爾類型TextUTF-8格式存儲的文本類型NullWritable空對象需要注意的是，這些數據類型的定義類都實現了WritableComparable接口。3.4.5MapReduce接口1.MapReduceWebUIMapReduceWebUI接口面向管理員?？梢栽陧撁嫔峡吹揭呀浲瓿傻乃蠱R-App執行過程中的統計信息，該頁面只支持讀，不支持寫。MapReduceWebUI的默認地址為http://JobHistoryServerIP:19888，可以查看MapReduce作業的歷史運行情況，如圖3-3所示。圖3-3MapReduce作業歷史情況2.MapReduceShellMapReduceShell接口面向MapReduce程序員。程序員通過Shell接口能夠向YARN集群提交MR-App，查看正在運行的MR-App，甚至可以終止正在運行的MR-App。MapReduceShell命令統一入口為“mapred”，語法格式如下：mapred[--configconfdir][--loglevelloglevel]COMMAND讀者需要注意的是，若$HADOOP_HOME/bin未加入到系統環境變量PATH中，則需要切換到Hadoop安裝目錄下，輸入“bin/mapred”。讀者可以使用“mapred-help”查看其幫助，命令“mapred”的具體用法和參數說明如圖3-4所示。圖3-4命令“mapred”用法MapReduceShell命令分為用戶命令和管理員命令。關于MapReduceShell命令的完整說明，讀者請參考官方網站/docs/r2.9.2/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapredCommands.html。3.MapReduceJavaAPIMapReduceJavaAPI接口面向Java開發工程師。程序員可以通過該接口編寫MR-App用戶層代碼MRApplicationBusinessLogic。MapReduce2.0應用程序包含3部分：MRv2框架中的MRAppMaster、MRClient再加上用戶編寫的MRApplicationBusinessLogic（Mapper類和Reduce類）。MR-App編寫步驟如下所示：（1）編寫MRApplicationBusinessLogic。自行編寫。（2）編寫MRApplicationMaster。無須編寫，Hadoop開發人員已編寫好MRAppMaster.java。（3）編寫MRApplicationClient。無須編寫，Hadoop開發人員已編寫好YARNRunner.java。其中，MRApplicationBusinessLogic編寫步驟如下：（1）確定<key,value>對。（2）定制輸入格式。（3）Mapper階段，其業務代碼需要繼承自org.apache.hadoop.mapreduce.Mapper類。（4）Reducer階段，其業務代碼需要繼承自org.apache.hadoop.mapreduce.Reducer類。（5）定制輸出格式。關于MapReduceAPI的完整說明，讀者請參考官方網站/docs/r2.9.2/api/index.html。3.5實驗步驟前文所講述的MapReduce入門案例WordCount統計的是單詞數量，而單詞本身屬于字面值，是比較容易計算的。本案例將會講解如何使用MapReduce統計對象中的某些屬性?！景咐?-1】以下是某個超市的結算記錄，從左往右各字段的含義依次是會員編號、結算時間、消費金額和用戶身份，樣例數據如下所示：242315 2019-10-15.18:20:10 32 會員984518 2019-10-15.18:21:02 167 會員226335 2019-10-15.18:21:54 233 非會員341665 2019-10-15.18:22:11 5 非會員273367 2019-10-15.18:23:07 361 非會員296223 2019-10-15.18:25:12 19 會員193363 2019-10-15.18:25:55 268 會員671512 2019-10-15.18:26:04 76 非會員596233 2019-10-15.18:27:42 82 非會員323444 2019-10-15.18:28:02 219 會員345672 2019-10-15.18:28:48 482 會員...現在需要完成以下任務：（1）編寫MapReduce程序計算會員和非會員的平均消費金額，并打包成jar文件提交Hadoop集群運行，查看運行結果。（2）在該MR-App運行過程中及運行結束后使用相關MapReduceShell命令，查看MR-App的運行狀態等信息。（3）在該MR-App運行過程中及運行結束后查看MapReduceWebUI界面，查看MR-App的運行狀態等信息。接下來從6個步驟完整闡述本案例的實現過程。3.5.1啟動Hadoop集群在主節點上依次執行以下3條命令啟動全分布模式Hadoop集群。start-dfs.shstart-yarn.shmr-jobhistory-daemon.shstarthistoryserver“start-dfs.sh”命令會在主節點上啟動NameNode和SecondaryNameNode服務，會在從節點上啟動DataNode服務；“start-yarn.sh”命令會在主節點上啟動ResourceManager服務，會在從節點上啟動NodeManager服務；“mr-jobhistory-daemon.shstarthistoryserver”命令會在主節點上啟動JobHistoryServer服務。3.5.2編寫并運行MapReduce程序在Hadoop集群主節點上搭建MapReduce開發環境Eclipse，具體過程請讀者參考本書2.6.4節，此處不再贅述。與2.6.5使用HDFSJavaAPI編程過程相同，在Eclipse項目MapReduceExample下建立新包com.xijing.mapreduce，模仿入門案例WordCount，編寫一個WordCount程序，最后打包成JAR形式并在Hadoop集群上運行該MR-App，查看運行結果。具體過程如下所示。1.在Eclipse中創建Java項目進入/usr/local/eclipse中通過可視化桌面打開EclipseIDE，默認的工作空間為“/home/xuluhui/eclipse-workspace”。選擇菜單『File』→『New』→『JavaProject』，創建Java項目“MapReduceExample”。2.在項目中導入所需JAR包為了編寫關于MapReduce應用程序，需要向Java項目中添加以下2個JAR包：hadoop-mapreduce-client-core-2.9.2.jar，MapReduce核心包，該包中包含了可以訪問MapReduce的JavaAPI，位于$HADOOP_HOME/share/hadoop/mapreduce下。hadoop-common-2.9.2.jar，由于需要對HDFS文件進行操作，所以還需要導入該包，該包位于$HADOOP_HOME/share/hadoop/common下。若不導入以上兩個JAR包，代碼將會出現錯誤。讀者可以參考2.6.5節步驟添加該應用程序編寫時所需的JAR包。3.在項目中新建包右鍵單擊項目“MapReduceExample”，從彈出的快捷菜單中選擇『New』→『Package』，創建包“com.xijing.mapreduce”。4.使用Java語言編寫MapReduce程序（1）編寫實體類右鍵單擊Java項目“MapReduceExample”中目錄“src”下的包“com.xijing.mapreduce”，從彈出的菜單中選擇『New』→『Class』，編寫封裝每個消費者記錄的實體類Customer，每個消費者至少包含了編號、消費金額和是否為會員等屬性。Customer.java源代碼如下所示。packagecom.xijing.mapreduce;importorg.apache.hadoop.io.Writable;importjava.io.DataInput;importjava.io.DataOutput;importjava.io.IOException;publicclassCustomerimplementsWritable{ //會員編號 privateStringid; //消費金額 privateintmoney; //0：非會員1：會員 privateintvip; publicCustomer(){ } publicCustomer(Stringid,intmoney,intvip){ this.id=id; this.money=money; this.vip=vip; } publicintgetMoney(){ returnmoney; } publicvoidsetMoney(intmoney){ this.money=money; } publicStringgetId(){ returnid; } publicvoidsetId(Stringid){ this.id=id; } publicintgetVip(){ returnvip; } publicvoidsetVip(intvip){ this.vip=vip; } //序列化 publicvoidwrite(DataOutputdataOutput)throwsIOException{ dataOutput.writeUTF(id); dataOutput.writeInt(money); dataOutput.writeInt(vip); } //反序列化（注意：各屬性的順序要和序列化保持一致） publicvoidreadFields(DataInputdataInput)throwsIOException{ this.id=dataInput.readUTF(); this.money=dataInput.readInt(); this.vip=dataInput.readInt(); } @Override publicStringtoString(){ returnthis.id+"\t"+this.money+"\t"+this.vip; }}由于本次統計的Customer對象需要在Hadoop集群中的多個節點之間傳遞數據，因此需要將Customer對象通過write（DataOutputdataOutput）方法進行序列化操作，并通過readFields（DataInputdataInput）進行反序列化操作。（2）編寫Mapper類右鍵單擊Java項目“MapReduceExample”中目錄“src”下的包“com.xijing.mapreduce”，從彈出的菜單中選擇『New』→『Class』，編寫Mapper類CustomerMapper。在Map階段讀取文本中的消費者記錄信息，并將消費者的各個屬性字段拆分讀取，然后根據會員情況，將消費者的消費金額輸出到MapReduce的下一個處理階段（即Shuffle），本案例Map階段輸出的數據形式是“會員（或非會員）,消費金額”。CustomerMapper.java源代碼如下所示。packagecom.xijing.mapreduce;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Mapper;importjava.io.IOException;publicclassCustomerMapperextendsMapper<LongWritable,Text,Text,IntWritable>{ @Override protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{ //將一行內容轉成string Stringline=value.toString(); //獲取各個顧客的消費數據 String[]fields=line.split("\t"); //獲取消費金額 intmoney=Integer.parseInt(fields[2]); //獲取會員情況 Stringvip=fields[3]; /* 輸出 Key：會員情況，value：消費金額 例如： 會員32 會員167 非會員233 非會員5 */ context.write(newText(vip),newIntWritable(money)); }}（3）編寫Reducer類右鍵單擊Java項目“MapReduceExample”中目錄“src”下的包“com.xijing.mapreduce”，從彈出的菜單中選擇『New』→『Class』，編寫Reducer類CustomerReducer。Map階段的輸出數據在經過shuffle階段混洗以后，就會傳遞給Reduce階段，本案例Reduce階段輸入的數據形式是“會員（或非會員）,[消費金額1,消費金額2,消費金額3,...]”。因此，與WordCount類似，只需要在Reduce階段累加會員或非會員的總消費金額就能完成本次任務。CustomerReducer.java源代碼如下所示。packagecom.xijing.mapreduce;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Reducer;importjava.io.IOException;publicclassCustomerReducerextendsReducer<Text,IntWritable,Text,LongWritable>{ @Override protectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{ //統計會員（或非會員）的個數 intvipCount=0; //總消費金額 longsumMoney=0; for(IntWritablemoney:values){ vipCount++; sumMoney+=money.get(); } //會員（或非會員）的平均消費金額 longavgMoney=sumMoney/vipCount; context.write(key,newLongWritable(avgMoney)); }}（4）編寫MapReduce程序驅動類右鍵單擊Java項目“MapReduceExample”中目錄“src”下的包“com.xijing.mapreduce”，從彈出的菜單中選擇『New』→『Class』，編寫MapReduce程序驅動類CustomerDriver。在編寫MapReduce程序時，程序的驅動類基本是相同的，因此，可以仿照之前的驅動類，編寫本次的MapReduce驅動類。CustomerDriver.java源代碼如下所示。packagecom.xijing.mapreduce;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;publicclassCustomerDriver{ publicstaticvoidmain(String[]args)throwsException{ Configurationconf=newConfiguration(); Jobjob=Job.getInstance(conf); job.setJarByClass(CustomerDriver.class); job.setMapperClass(CustomerMapper.class); job.setReducerClass(CustomerReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); FileInputFormat.setInputPaths(job,newPath(args[0])); FileOutputFormat.setOutputPath(job,newPath(args[1])); booleanresult=job.waitForCompletion(true); System.exit(result?0:1); }}最后，使用與WordCount程序相同的方法，將本程序打包成JAR包后就可以提交到Hadoop集群中運行了。執行結果就是會員與非會員的平均消費金額。5.將MapReduce程序打包成JAR包與實驗2相同，為了運行寫好的MapReduce程序，需要首先將程序打包成JAR包，具體打包成JAR包的過程此處不再贅述。編者將其保存在/home/xuluhui/eclipse-workspace/MapReduceExample下，命名為Customer.jar。該Java項目當前整體架構效果如圖3-5所示。圖3