實(shí)訓(xùn)3.2單詞計(jì)數(shù)1.實(shí)訓(xùn)目的基于MapReduce編程思想，編寫(xiě)WordCount程序。2.實(shí)訓(xùn)內(nèi)容理解MapReduce編程思想，掌握編寫(xiě)MapReduce版本W(wǎng)ordCount，了解該程序的執(zhí)行流程，結(jié)合執(zhí)行過(guò)程與結(jié)果，理解MapReduce的原理。3.實(shí)訓(xùn)要求以小組為單元進(jìn)行實(shí)訓(xùn)，每小組5人，小組自協(xié)商選一位組長(zhǎng)，由組長(zhǎng)安排和分配實(shí)訓(xùn)任務(wù)，具體參考實(shí)訓(xùn)操作環(huán)節(jié)。小組成員需要具備HDFS分布式存儲(chǔ)基礎(chǔ)相關(guān)知識(shí)，并確保具備前面的實(shí)訓(xùn)環(huán)境。4.準(zhǔn)備知識(shí)（1）MapReduce編程編寫(xiě)在Hadoop中依賴(lài)YARN框架執(zhí)行的MapReduce程序，大部分情況下只需要編寫(xiě)相應(yīng)的Map處理和Reduce處理過(guò)程的業(yè)務(wù)程序即可，因?yàn)榇蟛糠值拇a，Hadoop已經(jīng)幫我們寫(xiě)好了。所以編寫(xiě)一個(gè)MapReduce程序并不復(fù)雜，關(guān)鍵點(diǎn)在于掌握分布式的編程思想和方法，主要將計(jì)算過(guò)程分為以下五個(gè)步驟：①迭代。遍歷輸入數(shù)據(jù)，并將之解析成<Key,Value>鍵值對(duì)。②將輸入的<Key,Value>鍵值對(duì)映射（map）成新的<Key,Value>鍵值對(duì)。③依據(jù)Key對(duì)中間數(shù)據(jù)進(jìn)行分組。④以組為單位對(duì)數(shù)據(jù)進(jìn)行Reduce。⑤迭代。將最終產(chǎn)生的<Key,Value>鍵值對(duì)保存到輸出文件中。（2）JavaAPI解析①I(mǎi)nputFormat：用于描述輸入數(shù)據(jù)的格式，常用為T(mén)extInputFormat，其提供如下兩個(gè)功能：數(shù)據(jù)切分：按照某種策略將輸入數(shù)據(jù)切分成若干個(gè)數(shù)據(jù)塊，以便確定MapTask個(gè)數(shù)以及對(duì)應(yīng)的數(shù)據(jù)分片。為Map任務(wù)提供數(shù)據(jù)：給定某個(gè)數(shù)據(jù)分片，能將其解析成一個(gè)個(gè)的<Key,Value>鍵值對(duì)。②OutputFormat：用于描述輸出數(shù)據(jù)的格式，它能夠?qū)⒂脩?hù)提供的<Key,Value>鍵值對(duì)寫(xiě)入特定格式的文件中。③Mapper/Reducer：封裝了應(yīng)用程序的數(shù)據(jù)處理邏輯。④Writable：Hadoop自定義的序列化接口，實(shí)現(xiàn)該類(lèi)的接口可以用作MapReduce過(guò)程中的Value數(shù)據(jù)使用。⑤WritableComparable：在Writable基礎(chǔ)上繼承了Comparable接口，實(shí)現(xiàn)該類(lèi)的接口可以用作MapReduce過(guò)程中的Key數(shù)據(jù)使用。因?yàn)镵ey包含了比較和排序的操作。5.實(shí)訓(xùn)步驟本實(shí)訓(xùn)包含四大步驟：準(zhǔn)備運(yùn)行環(huán)境、準(zhǔn)備統(tǒng)計(jì)數(shù)據(jù)、編寫(xiě)MapReduce程序、打包并運(yùn)行代碼。查看運(yùn)行結(jié)果這幾個(gè)步驟，詳細(xì)如下：（1）準(zhǔn)備運(yùn)行環(huán)境啟動(dòng)三臺(tái)節(jié)點(diǎn)，然后在master中啟動(dòng)HDFS和YARN：start-dfs.shstart-yarn.sh（2）準(zhǔn)備統(tǒng)計(jì)數(shù)據(jù)新建待統(tǒng)計(jì)文件word.txt，并上傳數(shù)據(jù)到HDFS上vim/root/datas/word.txt內(nèi)容如下：hellohdfshadoophivehellomapreducehellosparksparkaibigdata將word.txt文件上傳到HDFS集群hdfsdfs-put/root/datas/word.txt/查看是否上傳成功，如圖3-所示則表示上傳成功。圖3-查看上傳結(jié)果（3）引入編寫(xiě)MapReduce程序相關(guān)的jar包本次實(shí)訓(xùn)繼續(xù)使用實(shí)訓(xùn)2.3的環(huán)境，但編寫(xiě)MapReduce程序需要引入相關(guān)的jar包，此時(shí)繼續(xù)引入hadoop-mapreduce-client-core.jar包。如圖3-所示。圖3-導(dǎo)入MapReduce依賴(lài)包（4）編寫(xiě)MapReduce程序編寫(xiě)MapReduce程序主要是編寫(xiě)Map類(lèi)和Reduce類(lèi)，其中Map過(guò)程需要繼承org.apache.hadoop.mapreduce包中Mapper類(lèi)，并重寫(xiě)其map方法；Reduce過(guò)程需要繼承org.apache.hadoop.mapreduce包中Reducer類(lèi)，并重寫(xiě)其reduce方法。在com.bigdata包（沒(méi)有則新建）下新建WordCount類(lèi)，添加以下代碼：packagecom.bigdata;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.TextInputFormat;importorg.apache.hadoop.mapreduce.lib.output.TextOutputFormat;importorg.apache.hadoop.mapreduce.lib.partition.HashPartitioner;importjava.io.IOException;importjava.util.StringTokenizer;publicclassWordCount{publicstaticclassTokenizerMapperextendsMapper<Object,Text,Text,IntWritable>{privatefinalstaticIntWritableone=newIntWritable(1);privateTextword=newText();//map方法，劃分一行文本，讀一個(gè)單詞寫(xiě)出一個(gè)<單詞,1>publicvoidmap(ObjectKey,Textvalue,Contextcontext)throwsIOException,InterruptedException{StringTokenizeritr=newStringTokenizer(value.toString());while(itr.hasMoreTokens()){word.set(itr.nextToken());context.write(word,one);//寫(xiě)出<單詞,1>}}}//定義reduce類(lèi)，對(duì)相同的單詞，把它們<K,VList>中的VList值全部相加publicstaticclassIntSumReducerextendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();publicvoidreduce(TextKey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritableval:values){sum+=val.get();//相當(dāng)于<Hello,1><Hello,1>，將兩個(gè)1相加}result.set(sum);context.write(Key,result);//寫(xiě)出這個(gè)單詞，和這個(gè)單詞出現(xiàn)次數(shù)<單詞，單詞出現(xiàn)次數(shù)>}} publicstaticvoidmain(String[]args)throwsException{//主方法，函數(shù)入口 //TODOAuto-generatedmethodstub Configurationconf=newConfiguration();//實(shí)例化配置文件類(lèi)Jobjob=newJob(conf,"WordCount");//實(shí)例化Job類(lèi)job.setInputFormatClass(TextInputFormat.class);//指定使用默認(rèn)輸入格式類(lèi)TextInputFormat.setInputPaths(job,args[0]);//設(shè)置待處理文件的位置job.setJarByClass(WordCount.class);//設(shè)置主類(lèi)名job.setMapperClass(TokenizerMapper.class);//指定使用上述自定義Map類(lèi)job.setCombinerClass(IntSumReducer.class);//指定開(kāi)啟Combiner函數(shù)job.setMapOutputKeyClass(Text.class);//指定Map類(lèi)輸出的<K,V>，K類(lèi)型job.setMapOutputValueClass(IntWritable.class);//指定Map類(lèi)輸出的<K,V>，V類(lèi)型job.setPartitionerClass(HashPartitioner.class);//指定使用默認(rèn)的HashPartitioner類(lèi)job.setReducerClass(IntSumReducer.class);//指定使用上述自定義Reduce類(lèi)job.setNumReduceTasks(Integer.parseInt(args[2]));//指定Reduce個(gè)數(shù)job.setOutputKeyClass(Text.class);//指定Reduce類(lèi)輸出的<K,V>,K類(lèi)型job.setOutputValueClass(Text.class);//指定Reduce類(lèi)輸出的<K,V>,V類(lèi)型job.setOutputFormatClass(TextOutputFormat.class);//指定使用默認(rèn)輸出格式類(lèi)TextOutputFormat.setOutputPath(job,newPath(args[1]));//設(shè)置輸出結(jié)果文件位置System.exit(job.waitForCompletion(true)?0:1);//提交任務(wù)并監(jiān)控任務(wù)狀態(tài) }}（5）打包并運(yùn)行代碼打包后的文件名為hadoop-project.jar，上傳到master節(jié)點(diǎn)的/root/jars目錄。主類(lèi)WordCount位于包c(diǎn)om.bigdata下，則可使用如下命令向YARN集群提交本應(yīng)用。yarnjarhadoop-project.jarcom.bigdata.WordCount/word.txt/wc_output1其中“yarn”為命令，“jar”為命令參數(shù)，后面緊跟打包后的代碼地址，“com.bigdata”為包名，“WordCount”為主類(lèi)名，第一個(gè)參數(shù)“/word.txt”為輸入文件在HDFS中的位置，第二個(gè)參數(shù)“/wc_output”為輸出文件在HDFS中的位置，第三個(gè)參數(shù)“1”表示需要統(tǒng)計(jì)成Reduce文件的個(gè)數(shù)。注意：“/word.txt”、“/wc_output”、“1”之間都有一個(gè)空格。同時(shí)需要啟動(dòng)HDFS和YARN。（6）實(shí)訓(xùn)結(jié)果程序運(yùn)行成功控制臺(tái)上的顯示內(nèi)容如圖3-15、圖3-16所示。圖3-15向YARN集群提交應(yīng)用圖3-16程序運(yùn)行結(jié)果在HDFS上查看結(jié)果，如圖3-17、圖3-18所示。hdfsdfs-ls/wc_output圖3-17查看/wc_output目錄下的目錄和文件hdfsdfs-cat/wc_output/part-r-00000圖3-18HDFS上顯示結(jié)果6.實(shí)訓(xùn)總結(jié)實(shí)訓(xùn)基于MapReduce編程思想，編寫(xiě)MRJava版