要統計的文件的文件名為hello

hello中的內容如下

hello　　you

hello　　me

通過MapReduce程序統計出文件中的各個單詞出現了幾次.(兩個單詞之間通過tab鍵進行的分割)

1 import java.io.IOException; 2 3 import mapreduce.WordCountApp.WordCountMapper.WordCountReducer; 4 5 import org.apache.hadoop.conf.Configuration; 6 import org.apache.hadoop.fs.Path; 7 import org.apache.hadoop.io.LongWritable; 8 import org.apache.hadoop.io.Text; 9 import org.apache.hadoop.mapreduce.Job; 10 import org.apache.hadoop.mapreduce.Mapper; 11 import org.apache.hadoop.mapreduce.Reducer; 12 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 13 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 14 15 /** 16 * 以文本 17 * hello you 18 * hello me 19 * 為例子. 20 * map方法調用了兩次,因為有兩行 21 * k2 v2 鍵值對的數量有幾個? 22 * 有4個.有四個單詞. 23 * 24 * 會產生幾個分組? 25 * 產生3個分組. 26 * 有3個不同的單詞. 27 * 28 */ 29 public class WordCountApp { 30 public static void main(String[] args) throws Exception { 31 //程序在這里運行,要有驅動. 32 Configuration conf = new Configuration(); 33 Job job = Job.getInstance(conf,WordCountApp.class.getSimpleName()); 34 35 //我們運行此程序通過運行jar包來執行.一定要有這句話. 36 job.setJarByClass(WordCountApp.class); 37 38 FileInputFormat.setInputPaths(job,args[0]); 39 40 job.setMapperClass(WordCountMapper.class);//設置Map類 41 job.setMapOutputKeyClass(Text.class);//設置Map的key 42 job.setMapOutputValueClass(LongWritable.class);//設置Map的value 43 job.setReducerClass(WordCountReducer.class);//設置Reduce的類 44 job.setOutputKeyClass(Text.class);//設置Reduce的key Reduce這個地方只有輸出的參數可以設置. 方法名字也沒有Reduce關鍵字區別于Map 45 job.setOutputValueClass(LongWritable.class);//設置Reduce的value. 46 47 FileOutputFormat.setOutputPath(job, new Path(args[1])); 48 job.waitForCompletion(true);//表示結束了才退出,不結束不退出 49 } 50 /** 51 * 4個泛型的意識 52 * 第一個是LongWritable,固定就是這個類型,表示每一行單詞的起始位置(單位是字節) 53 * 第二個是Text,表示每一行的文本內容. 54 * 第三個是Text,表示單詞 55 * 第四個是LongWritable,表示單詞的出現次數 56 */ 57 public static class WordCountMapper extends Mapper<LongWritable, Text, Text ,LongWritable>{ 58 Text k2 = new Text(); 59 LongWritable v2 = new LongWritable(); 60 //增加一個計數器,這個Map調用幾次就輸出對應的次數. 61 int counter = 0; 62 63 64 /** 65 * key和value表示輸入的信息 66 * 每一行文本調用一次map函數 67 */ 68 @Override 69 protected void map(LongWritable key, Text value,Mapper<LongWritable, Text, Text, LongWritable>.Context context) 70 throws IOException, InterruptedException { 71 counter = counter + 1; 72 System.out.println("mapper 調用的次數:" + counter); 73 //這個map方法中的Mapper的各個泛型和上面的意識是一樣的,分別代表的是k1,v1,k2,v2 74 String line = value.toString(); 75 System.out.println(String.format("<k1,v1>的值<"+key.get()+","+line+">")); 76 String[] splited = line.split("\t"); 77 for (String word : splited) { 78 k2.set(word); 79 v2.set(1); 80 System.out.println(String.format("<k2,v2>的值<"+k2.toString()+","+v2.get()+">")); 81 context.write(k2, v2);//通過context對象寫出去. 82 } 83 } 84 /** 85 * 這個地方的四個泛型的意思 86 * 前兩個泛型是對應的Map方法的后兩個泛型. 87 * Map的輸出對應的是Reduce的輸入. 88 * 第一個Text是單詞 89 * 第二個LongWritable是單詞對應的次數 90 * 我們想輸出的也是單詞 和 次數 91 * 所以第三個和第四個的類型和第一和第二個的一樣 92 * 93 * 分組指的是把相同key2的value2放到一個集合中 94 * 95 */ 96 public static class WordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable>{ 97 LongWritable v3 = new LongWritable(); 98 //增加一個計數器,這個Reduce調用幾次就輸出對應的次數. 99 int counter = 0; 100 101 /** 102 * 每一個分組調用一次reduce函數 103 * 過來的k2 分別是hello you me 104 * 105 */ 106 @Override 107 protected void reduce(Text key2, Iterable<LongWritable> value2Iterable,Reducer<Text, LongWritable, Text, 108 LongWritable>.Context context) 109 throws IOException, InterruptedException { 110 counter = counter + 1; 111 System.out.println("reducer 調用的次數:" + counter); 112 //第一個參數是單詞,第二個是可迭代的集合. 為什么上面的LongWritable類型的對象value2變成了一個可以迭代的結合參數? 113 //因為分組指的是把相同key2的value2放到一個集合中 114 long sum = 0L; 115 for (LongWritable value2 : value2Iterable) { 116 System.out.println(String.format("<k2,v2>的值<"+key2.toString()+","+value2.toString()+">")); 117 sum += value2.get(); //這個value2是LongWritable類型的,不能進行+= 操作,要用get()得到其對應的java基本類型. 118 //sum表示單詞k2 在整個文本中的出現次數. 119 } 120 v3.set(sum); 121 context.write(key2, v3); 122 System.out.println(String.format("<k3,v3>的值<"+key2.toString()+","+v3.get()+">")); 123 } 124 } 125 } 126 }

?

通過運行Yarn集群查看Map日志得到的輸出結果:?

查看Reduce日志產看到的輸出結果:

?

//============================================================================

以下程序是之前的寫的:注釋更加詳細:

1 /* 2 * 一個hello文件內容如下: 3 * hello you 4 * hello me 5 */ 6 import java.io.IOException; 7 8 import org.apache.hadoop.conf.Configuration; 9 import org.apache.hadoop.fs.Path; 10 import org.apache.hadoop.io.LongWritable; 11 import org.apache.hadoop.io.Text; 12 import org.apache.hadoop.mapreduce.Job; 13 import org.apache.hadoop.mapreduce.Mapper; 14 import org.apache.hadoop.mapreduce.Reducer; 15 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 16 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 17 18 public class WordCountApp { 19 public static void main(String[] args) throws Exception { 20 // 在main方法寫驅動程序,把Map函數和Reduce函數組織在一起. 21 // 搞一個對象把Map對象和Reduce對象都放在這個對象中,我們把這個對象稱作Job 22 // 兩個形參,一個是Configuration對象,一個是Job的名稱,這樣獲得了一個Job對象; 23 Job job = Job.getInstance(new Configuration(), 24 WordCountApp.class.getSimpleName()); 25 // 對這個job進行設置 26 job.setJarByClass(WordCountApp.class);// 通過這個設置可以讓框架識別你寫的代碼 27 28 job.setMapperClass(MyMapper.class);// 把自定義的Map類放到job中 29 job.setMapOutputKeyClass(Text.class);// 定義Map的key的輸出類型,Map的輸出是<hello,2> 30 job.setMapOutputValueClass(LongWritable.class);// 定義Map的value的輸出類型 31 32 job.setReducerClass(MyReducer.class);// 把自定義的Reducer類放到job中 33 job.setOutputKeyClass(Text.class);// 因為Reduce的輸出是最終的數據,Reduce的輸出是<hello,2> 34 // 所以這個方法名中沒有像Map對應的放發一樣帶有Reduce,直接就是setOutputKeyClass 35 job.setOutputValueClass(LongWritable.class);// 定義reduce的value輸出 36 37 FileInputFormat.setInputPaths(job, args[0]);// 輸入指定:傳入一個job地址. 38 // 這個args[0] 就是新地址,"hdfs://192.168.0.170/hello" 39 FileOutputFormat.setOutputPath(job, new Path(args[1])); 40 // 輸出指定 41 // 指定輸入和輸出路徑可以通過在這里寫死的方式,也可以通過main函數參數的形式 42 // 分別是args[0]和args[1] 43 44 // 把job上傳到yarn平臺上. 45 job.waitForCompletion(true); 46 } 47 48 /* 49 * 對于<k1,v1>而言,每一行產生一個<k1,v1>對,<k1,v1>表示<行的起始位置,行的文本內容> 50 * 就本例而言map函數總共調用兩次,因為總共只有兩行. 51 * 正對要統計的文本內容可以知道總共兩行,總共會調用兩次Map函數對應產生的<k1,v1>分別是<0,hello you> 52 * 和第二個<k1,v1>是<10,hello me> 53 */ 54 private static class MyMapper extends 55 Mapper<LongWritable, Text, Text, LongWritable> { 56 // 這個Mapper的泛型參數是<KEYIN,VALUEIN,KEYOUT,VALUEOUT> 分別對應的是k1,v1,k2,v2 57 // 我們如下講的k1,v1的類型是固定的. 58 // 就本例而言，map函數會被調用2次,因為總共文本文件就只有兩行. 59 60 //要定義輸出的k2和v2.本案例中可以分析出<k2,v2>是對文本內容的統計<hello,1><hello,1><you,1><me,1> 61 //而且<k2,v2>的內容是和<k3,v3>中的內容是一樣的. 62 Text k2 = new Text(); 63 LongWritable v2 = new LongWritable(); 64 //重寫父類Mapper中的map方法 65 @Override 66 protected void map(LongWritable key, Text value, 67 Mapper<LongWritable, Text, Text, LongWritable>.Context context) 68 throws IOException, InterruptedException { 69 //通過代碼或者案例分析就可以知道k1其實沒有什么用出的. 70 String line = value.toString(); 71 String[] splited = line.split("\t");//根據制表分隔符機進行拆分.hello和me,you之間是一個制表分隔符. 72 for (String word : splited) { 73 k2.set(word); 74 v2.set(1); 75 context.write(k2, v2); 76 //用context把k2,v2寫出去,框架會寫,不用我們去管. 77 } 78 } 79 } 80 81 private static class MyReducer extends 82 Reducer<Text, LongWritable, Text, LongWritable> { 83 //這個例子中的<k2,v2>和<k3,v3>中的k是一樣的,所以這里,k2當做k3了. 84 LongWritable v3 = new LongWritable(); 85 @Override 86 protected void reduce(Text k2, Iterable<LongWritable> v2s, 87 Reducer<Text, LongWritable, Text, LongWritable>.Context context) 88 throws IOException, InterruptedException { 89 //Reduce是對上面Map中的結果進行匯總的. 90 //上面拆分出來的<k2,v2>是<hello,1><hello,1><you,1><me,1>Reduce方法中就要對其進行匯總. 91 long sum = 0L; 92 for(LongWritable v2:v2s){ 93 sum = sum +v2.get();//sum是long類型,v2是LongWritable類型 94 //LongWritable類型轉換成long類型用get()方法. 95 //sum的值表示單詞在整個文件中出現的中次數. 96 } 97 v3.set(sum); 98 context.write(k2,v3); 99 } 100 } 101 102 }

?//===============================================================================

查看日志的時候,代碼中的System.out.println()對于Java程序輸出到控制臺,但是這個地方是把Java類打成Jar包,

放到集群中去通過命令執行的.

輸出通過日志查看的.

上面對應的Log Type:stdout?

stdout:stdout（Standardoutput）標準輸出

本文轉自SummerChill博客園博客，原文鏈接：http://www.cnblogs.com/DreamDrive/p/5492572.html，如需轉載請自行聯系原作者

總結

以上是生活随笔為你收集整理的关于MapReduce单词统计的例子:的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。