要統(tǒng)計(jì)的文件的文件名為hello

hello中的內(nèi)容如下

hello　　you

hello　　me

通過(guò)MapReduce程序統(tǒng)計(jì)出文件中的各個(gè)單詞出現(xiàn)了幾次.(兩個(gè)單詞之間通過(guò)tab鍵進(jìn)行的分割)

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方法調(diào)用了兩次,因?yàn)橛袃尚? 21 * k2 v2 鍵值對(duì)的數(shù)量有幾個(gè)? 22 * 有4個(gè).有四個(gè)單詞. 23 * 24 * 會(huì)產(chǎn)生幾個(gè)分組? 25 * 產(chǎn)生3個(gè)分組. 26 * 有3個(gè)不同的單詞. 27 * 28 */ 29 public class WordCountApp { 30 public static void main(String[] args) throws Exception { 31 //程序在這里運(yùn)行,要有驅(qū)動(dòng). 32 Configuration conf = new Configuration(); 33 Job job = Job.getInstance(conf,WordCountApp.class.getSimpleName()); 34 35 //我們運(yùn)行此程序通過(guò)運(yùn)行jar包來(lái)執(zhí)行.一定要有這句話. 36 job.setJarByClass(WordCountApp.class); 37 38 FileInputFormat.setInputPaths(job,args[0]); 39 40 job.setMapperClass(WordCountMapper.class);//設(shè)置Map類(lèi) 41 job.setMapOutputKeyClass(Text.class);//設(shè)置Map的key 42 job.setMapOutputValueClass(LongWritable.class);//設(shè)置Map的value 43 job.setReducerClass(WordCountReducer.class);//設(shè)置Reduce的類(lèi) 44 job.setOutputKeyClass(Text.class);//設(shè)置Reduce的key Reduce這個(gè)地方只有輸出的參數(shù)可以設(shè)置. 方法名字也沒(méi)有Reduce關(guān)鍵字區(qū)別于Map 45 job.setOutputValueClass(LongWritable.class);//設(shè)置Reduce的value. 46 47 FileOutputFormat.setOutputPath(job, new Path(args[1])); 48 job.waitForCompletion(true);//表示結(jié)束了才退出,不結(jié)束不退出 49 } 50 /** 51 * 4個(gè)泛型的意識(shí) 52 * 第一個(gè)是LongWritable,固定就是這個(gè)類(lèi)型,表示每一行單詞的起始位置(單位是字節(jié)) 53 * 第二個(gè)是Text,表示每一行的文本內(nèi)容. 54 * 第三個(gè)是Text,表示單詞 55 * 第四個(gè)是LongWritable,表示單詞的出現(xiàn)次數(shù) 56 */ 57 public static class WordCountMapper extends Mapper<LongWritable, Text, Text ,LongWritable>{ 58 Text k2 = new Text(); 59 LongWritable v2 = new LongWritable(); 60 //增加一個(gè)計(jì)數(shù)器,這個(gè)Map調(diào)用幾次就輸出對(duì)應(yīng)的次數(shù). 61 int counter = 0; 62 63 64 /** 65 * key和value表示輸入的信息 66 * 每一行文本調(diào)用一次map函數(shù) 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 調(diào)用的次數(shù):" + counter); 73 //這個(gè)map方法中的Mapper的各個(gè)泛型和上面的意識(shí)是一樣的,分別代表的是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);//通過(guò)context對(duì)象寫(xiě)出去. 82 } 83 } 84 /** 85 * 這個(gè)地方的四個(gè)泛型的意思 86 * 前兩個(gè)泛型是對(duì)應(yīng)的Map方法的后兩個(gè)泛型. 87 * Map的輸出對(duì)應(yīng)的是Reduce的輸入. 88 * 第一個(gè)Text是單詞 89 * 第二個(gè)LongWritable是單詞對(duì)應(yīng)的次數(shù) 90 * 我們想輸出的也是單詞 和 次數(shù) 91 * 所以第三個(gè)和第四個(gè)的類(lèi)型和第一和第二個(gè)的一樣 92 * 93 * 分組指的是把相同key2的value2放到一個(gè)集合中 94 * 95 */ 96 public static class WordCountReducer extends Reducer<Text, LongWritable, Text, LongWritable>{ 97 LongWritable v3 = new LongWritable(); 98 //增加一個(gè)計(jì)數(shù)器,這個(gè)Reduce調(diào)用幾次就輸出對(duì)應(yīng)的次數(shù). 99 int counter = 0; 100 101 /** 102 * 每一個(gè)分組調(diào)用一次reduce函數(shù) 103 * 過(guò)來(lái)的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 調(diào)用的次數(shù):" + counter); 112 //第一個(gè)參數(shù)是單詞,第二個(gè)是可迭代的集合. 為什么上面的LongWritable類(lèi)型的對(duì)象value2變成了一個(gè)可以迭代的結(jié)合參數(shù)? 113 //因?yàn)榉纸M指的是把相同key2的value2放到一個(gè)集合中 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(); //這個(gè)value2是LongWritable類(lèi)型的,不能進(jìn)行+= 操作,要用get()得到其對(duì)應(yīng)的java基本類(lèi)型. 118 //sum表示單詞k2 在整個(gè)文本中的出現(xiàn)次數(shù). 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 }

?

通過(guò)運(yùn)行Yarn集群查看Map日志得到的輸出結(jié)果:?

查看Reduce日志產(chǎn)看到的輸出結(jié)果:

?

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

以下程序是之前的寫(xiě)的:注釋更加詳細(xì):

1 /* 2 * 一個(gè)hello文件內(nèi)容如下: 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方法寫(xiě)驅(qū)動(dòng)程序,把Map函數(shù)和Reduce函數(shù)組織在一起. 21 // 搞一個(gè)對(duì)象把Map對(duì)象和Reduce對(duì)象都放在這個(gè)對(duì)象中,我們把這個(gè)對(duì)象稱(chēng)作Job 22 // 兩個(gè)形參,一個(gè)是Configuration對(duì)象,一個(gè)是Job的名稱(chēng),這樣獲得了一個(gè)Job對(duì)象; 23 Job job = Job.getInstance(new Configuration(), 24 WordCountApp.class.getSimpleName()); 25 // 對(duì)這個(gè)job進(jìn)行設(shè)置 26 job.setJarByClass(WordCountApp.class);// 通過(guò)這個(gè)設(shè)置可以讓框架識(shí)別你寫(xiě)的代碼 27 28 job.setMapperClass(MyMapper.class);// 把自定義的Map類(lèi)放到j(luò)ob中 29 job.setMapOutputKeyClass(Text.class);// 定義Map的key的輸出類(lèi)型,Map的輸出是<hello,2> 30 job.setMapOutputValueClass(LongWritable.class);// 定義Map的value的輸出類(lèi)型 31 32 job.setReducerClass(MyReducer.class);// 把自定義的Reducer類(lèi)放到j(luò)ob中 33 job.setOutputKeyClass(Text.class);// 因?yàn)镽educe的輸出是最終的數(shù)據(jù),Reduce的輸出是<hello,2> 34 // 所以這個(gè)方法名中沒(méi)有像Map對(duì)應(yīng)的放發(fā)一樣帶有Reduce,直接就是setOutputKeyClass 35 job.setOutputValueClass(LongWritable.class);// 定義reduce的value輸出 36 37 FileInputFormat.setInputPaths(job, args[0]);// 輸入指定:傳入一個(gè)job地址. 38 // 這個(gè)args[0] 就是新地址,"hdfs://192.168.0.170/hello" 39 FileOutputFormat.setOutputPath(job, new Path(args[1])); 40 // 輸出指定 41 // 指定輸入和輸出路徑可以通過(guò)在這里寫(xiě)死的方式,也可以通過(guò)main函數(shù)參數(shù)的形式 42 // 分別是args[0]和args[1] 43 44 // 把job上傳到y(tǒng)arn平臺(tái)上. 45 job.waitForCompletion(true); 46 } 47 48 /* 49 * 對(duì)于<k1,v1>而言,每一行產(chǎn)生一個(gè)<k1,v1>對(duì),<k1,v1>表示<行的起始位置,行的文本內(nèi)容> 50 * 就本例而言map函數(shù)總共調(diào)用兩次,因?yàn)榭偣仓挥袃尚? 51 * 正對(duì)要統(tǒng)計(jì)的文本內(nèi)容可以知道總共兩行,總共會(huì)調(diào)用兩次Map函數(shù)對(duì)應(yīng)產(chǎn)生的<k1,v1>分別是<0,hello you> 52 * 和第二個(gè)<k1,v1>是<10,hello me> 53 */ 54 private static class MyMapper extends 55 Mapper<LongWritable, Text, Text, LongWritable> { 56 // 這個(gè)Mapper的泛型參數(shù)是<KEYIN,VALUEIN,KEYOUT,VALUEOUT> 分別對(duì)應(yīng)的是k1,v1,k2,v2 57 // 我們?nèi)缦轮v的k1,v1的類(lèi)型是固定的. 58 // 就本例而言，map函數(shù)會(huì)被調(diào)用2次,因?yàn)榭偣参谋疚募椭挥袃尚? 59 60 //要定義輸出的k2和v2.本案例中可以分析出<k2,v2>是對(duì)文本內(nèi)容的統(tǒng)計(jì)<hello,1><hello,1><you,1><me,1> 61 //而且<k2,v2>的內(nèi)容是和<k3,v3>中的內(nèi)容是一樣的. 62 Text k2 = new Text(); 63 LongWritable v2 = new LongWritable(); 64 //重寫(xiě)父類(lèi)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 //通過(guò)代碼或者案例分析就可以知道k1其實(shí)沒(méi)有什么用出的. 70 String line = value.toString(); 71 String[] splited = line.split("\t");//根據(jù)制表分隔符機(jī)進(jìn)行拆分.hello和me,you之間是一個(gè)制表分隔符. 72 for (String word : splited) { 73 k2.set(word); 74 v2.set(1); 75 context.write(k2, v2); 76 //用context把k2,v2寫(xiě)出去,框架會(huì)寫(xiě),不用我們?nèi)ス? 77 } 78 } 79 } 80 81 private static class MyReducer extends 82 Reducer<Text, LongWritable, Text, LongWritable> { 83 //這個(gè)例子中的<k2,v2>和<k3,v3>中的k是一樣的,所以這里,k2當(dāng)做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是對(duì)上面Map中的結(jié)果進(jìn)行匯總的. 90 //上面拆分出來(lái)的<k2,v2>是<hello,1><hello,1><you,1><me,1>Reduce方法中就要對(duì)其進(jìn)行匯總. 91 long sum = 0L; 92 for(LongWritable v2:v2s){ 93 sum = sum +v2.get();//sum是long類(lèi)型,v2是LongWritable類(lèi)型 94 //LongWritable類(lèi)型轉(zhuǎn)換成long類(lèi)型用get()方法. 95 //sum的值表示單詞在整個(gè)文件中出現(xiàn)的中次數(shù). 96 } 97 v3.set(sum); 98 context.write(k2,v3); 99 } 100 } 101 102 }

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

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

放到集群中去通過(guò)命令執(zhí)行的.

輸出通過(guò)日志查看的.

上面對(duì)應(yīng)的Log Type:stdout?

stdout:stdout（Standardoutput）標(biāo)準(zhǔn)輸出

本文轉(zhuǎn)自SummerChill博客園博客，原文鏈接：http://www.cnblogs.com/DreamDrive/p/5492572.html，如需轉(zhuǎn)載請(qǐng)自行聯(lián)系原作者

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。