實訓5 單詞計數(shù)

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