### 5.1 實驗目的 基于MapReduce思想，編寫WordCount程序。 ### 5.2 實驗要求 1． 理解MapReduce編程思想； 2． 會編寫MapReduce版本WordCount； 3． 會執行該程序； 4． 自行分析執行過程。 ### 5.3 實驗原理 MapReduce是一種計算模型，簡單的說就是將大批量的工作（數據）分解（MAP）執行，然后再將結果合并成最終結果（REDUCE）。這樣做的好處是可以在任務被分解后，可以通過大量機器進行并行計算，減少整個操作的時間。 適用范圍：數據量大，但是數據種類小可以放入內存。 基本原理及要點：將數據交給不同的機器去處理，數據劃分，結果歸約。 理解MapReduce和Yarn：在新版Hadoop中，Yarn作為一個資源管理調度框架，是Hadoop下MapReduce程序運行的生存環境。其實MapRuduce除了可以運行Yarn框架下，也可以運行在諸如Mesos，Corona之類的調度框架上，使用不同的調度框架，需要針對Hadoop做不同的適配。 一個完成的MapReduce程序在Yarn中執行過程如下： （1）ResourcManager JobClient向ResourcManager提交一個job。 （2）ResourcManager向Scheduler請求一個供MRAppMaster運行的container，然后啟動它。 （3）MRAppMaster啟動起來后向ResourcManager注冊。 （4）ResourcManagerJobClient向ResourcManager獲取到MRAppMaster相關的信息，然后直接與MRAppMaster進行通信。 （5）MRAppMaster算splits并為所有的map構造資源請求。 （6）MRAppMaster做一些必要的MR OutputCommitter的準備工作。 （7）MRAppMaster向RM(Scheduler)發起資源請求，得到一組供map/reduce task運行的container，然后與NodeManager一起對每一個container執行一些必要的任務，包括資源本地化等。 （8）MRAppMaster 監視運行著的task 直到完成，當task失敗時，申請新的container運行失敗的task。 （9）當每個map/reduce task完成后，MRAppMaster運行MR OutputCommitter的cleanup 代碼，也就是進行一些收尾工作。 （10）當所有的map/reduce完成后，MRAppMaster運行OutputCommitter的必要的job commit或者abort APIs。 （11）MRAppMaster退出。 #### 5.3.1 MapReduce編程 編寫在Hadoop中依賴Yarn框架執行的MapReduce程序，并不需要自己開發MRAppMaster和YARNRunner，因為Hadoop已經默認提供通用的YARNRunner和MRAppMaster程序， 大部分情況下只需要編寫相應的Map處理和Reduce處理過程的業務程序即可。 編寫一個MapReduce程序并不復雜，關鍵點在于掌握分布式的編程思想和方法，主要將計算過程分為以下五個步驟： （1）迭代。遍歷輸入數據，并將之解析成key/value對。 （2）將輸入key/value對映射(map)成另外一些key/value對。 （3）依據key對中間數據進行分組(grouping)。 （4）以組為單位對數據進行歸約(reduce)。 （5）迭代。將最終產生的key/value對保存到輸出文件中。 #### 5.3.2 Java API解析 （1）InputFormat：用于描述輸入數據的格式，常用的為TextInputFormat提供如下兩個功能： 數據切分： 按照某個策略將輸入數據切分成若干個split，以便確定Map Task個數以及對應的split。 為Mapper提供數據：給定某個split，能將其解析成一個個key/value對。 （2）OutputFormat：用于描述輸出數據的格式，它能夠將用戶提供的key/value對寫入特定格式的文件中。 （3）Mapper/Reducer: Mapper/Reducer中封裝了應用程序的數據處理邏輯。 （4）Writable:Hadoop自定義的序列化接口。實現該類的接口可以用作MapReduce過程中的value數據使用。 （5）WritableComparable：在Writable基礎上繼承了Comparable接口，實現該類的接口可以用作MapReduce過程中的key數據使用。（因為key包含了比較排序的操作）。 ### 5.4 實驗步驟 本實驗主要分為，確認前期準備，編寫MapReduce程序，打包提交代碼。查看運行結果這幾個步驟，詳細如下： #### 5.4.1 啟動Hadoop 執行命令啟動前面實驗部署好的Hadoop系統。 ~~~ [root@master ~]# cd /usr/cstor/hadoop/ [root@master hadoop]# sbin/start-all.sh ~~~ #### 5.4.2 驗證HDFS上沒有wordcount的文件夾 ~~~ [root@client ~]# cd /usr/cstor/hadoop/ [root@client hadoop]# bin/hadoop fs -ls / #查看HDFS上根目錄文件 / ~~~ 此時HDFS上應該是沒有wordcount文件夾。 #### 5.4.3 上傳數據文件到HDFS ~~~ [root@client ~]# cd /usr/cstor/hadoop/ [root@client hadoop]# bin/hadoop fs -put /root/data/5/word / ~~~ #### 5.4.4 編寫MapReduce程序 主要編寫Map和Reduce類，其中Map過程需要繼承org.apache.hadoop.mapreduce包中Mapper類，并重寫其map方法；Reduce過程需要繼承org.apache.hadoop.mapreduce包中Reduce類，并重寫其reduce方法。 ~~~ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner; import java.io.IOException; import java.util.StringTokenizer; public class WordCount { public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); //map方法，劃分一行文本，讀一個單詞寫出一個<單詞,1> public void map(Object key, Text value, Context context)throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one);//寫出<單詞,1> }}} //定義reduce類，對相同的單詞，把它們<K,VList>中的VList值全部相加 public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get();//相當于<Hello,1><Hello,1>，將兩個1相加 } result.set(sum); context.write(key, result);//寫出這個單詞，和這個單詞出現次數<單詞，單詞出現次數> }} public static void main(String[] args) throws Exception {//主方法，函數入口 Configuration conf = new Configuration(); //實例化配置文件類 Job job = new Job(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函數 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個數 job.setOutputKeyClass(Text.class); //指定Reduce類輸出的<K,V>,K類型 job.setOutputValueClass(Text.class); //指定Reduce類輸出的<K,V>,V類型 job.setOutputFormatClass(TextOutputFormat.class); //指定使用默認輸出格式類 TextOutputFormat.setOutputPath(job, new Path(args[1])); //設置輸出結果文件位置 System.exit(job.waitForCompletion(true) ? 0 : 1); //提交任務并監控任務狀態 } } ~~~ #### 5.3.5 使用Eclipse開發工具將該代碼打包 假定打包后的文件名為hdpAction.jar，主類WordCount位于包njupt下，則可使用如下命令向YARN集群提交本應用。 ~~~ [root@client ~]# ./yarn jar hdpAction.jar njupt.WordCount /word /wordcount 1 ~~~ 其中“yarn”為命令，“jar”為命令參數，后面緊跟打包后的代碼地址，“njupt”為包名，“WordCount”為主類名，“/word”為輸入文件在HDFS中的位置，/wordcount為輸出文件在HDFS中的位置。 ### 5.5 實驗結果 #### 5.5.1 程序運行成功控制臺上的顯示內容 如圖5-1所示：  圖5-1 提交wordcount #### 5.5.2 在HDFS上查看結果 如圖5-2所示：