我的理解：
一個(gè)有三行文本的文件進(jìn)行MapReduce操作。
讀取第一行Hello World Bye World ，分割單詞形成Map。
<Hello,1> <World,1> <Bye,1> <World,1>
讀取第二行Hello Hadoop Bye Hadoop ，分割單詞形成Map。
<Hello,1> <Hadoop,1> <Bye,1> <Hadoop,1>
讀取第三行Bye Hadoop Hello Hadoop，分割單詞形成Map。
<Bye,1> <Hadoop,1> <Hello,1> <Hadoop,1>
我的理解：
經(jīng)過(guò)進(jìn)一步處理(combiner),將形成的Map根據相同的key組合成value數組。
<Bye,1,1,1> <Hadoop,1,1,1,1> <Hello,1,1,1> <World,1,1>
循環(huán)執行Reduce(K,V[])，分別統計每個(gè)單詞出現的次數。
<Bye,3> <Hadoop,4> <Hello,3> <World,2>
package org.apache.hadoop.examples;import java.io.IOException;import java.util.StringTokenizer;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.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.util.GenericOptionsParser;/** * * 描述：WordCount explains by York * @author Hadoop Dev Group */publicclass WordCount { /** * 建立Mapper類(lèi)TokenizerMapper繼承自泛型類(lèi)Mapper * Mapper類(lèi):實(shí)現了Map功能基類(lèi) * Mapper接口： * WritableComparable接口：實(shí)現WritableComparable的類(lèi)可以相互比較。所有被用作key的類(lèi)應該實(shí)現此接口。 * Reporter 則可用于報告整個(gè)應用的運行進(jìn)度，本例中未使用。 * */ publicstaticclass TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ /** * IntWritable, Text 均是 Hadoop 中實(shí)現的用于封裝 Java 數據類(lèi)型的類(lèi)，這些類(lèi)實(shí)現了WritableComparable接口， * 都能夠被串行化從而便于在分布式環(huán)境中進(jìn)行數據交換，你可以將它們分別視為int,String 的替代品。 * 聲明one常量和word用于存放單詞的變量 */ privatefinalstatic IntWritable one =new IntWritable(1); private Text word =new Text(); /** * Mapper中的map方法： * void map(K1 key, V1 value, Context context) * 映射一個(gè)單個(gè)的輸入k/v對到一個(gè)中間的k/v對 * 輸出對不需要和輸入對是相同的類(lèi)型，輸入對可以映射到0個(gè)或多個(gè)輸出對。 * Context：收集Mapper輸出的<k,v>對。 * Context的write(k, v)方法:增加一個(gè)(k,v)對到context * 程序員主要編寫(xiě)Map和Reduce函數.這個(gè)Map函數使用StringTokenizer函數對字符串進(jìn)行分隔,通過(guò)write方法把單詞存入word中 * write方法存入(單詞,1)這樣的二元組到context中 */ publicvoid 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); } } } publicstaticclass IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result =new IntWritable(); /** * Reducer類(lèi)中的reduce方法： * void reduce(Text key, Iterable<IntWritable> values, Context context) * 中k/v來(lái)自于map函數中的context,可能經(jīng)過(guò)了進(jìn)一步處理(combiner),同樣通過(guò)context輸出 */ publicvoid reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum =0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } publicstaticvoid main(String[] args) throws Exception { /** * Configuration：map/reduce的j配置類(lèi)，向hadoop框架描述map-reduce執行的工作 */ Configuration conf =new Configuration(); String[] otherArgs =new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length !=2) { System.err.println("Usage: wordcount <in> <out>"); System.exit(2); } Job job =new Job(conf, "word count"); //設置一個(gè)用戶(hù)定義的job名稱(chēng) job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); //為job設置Mapper類(lèi) job.setCombinerClass(IntSumReducer.class); //為job設置Combiner類(lèi) job.setReducerClass(IntSumReducer.class); //為job設置Reducer類(lèi) job.setOutputKeyClass(Text.class); //為job的輸出數據設置Key類(lèi) job.setOutputValueClass(IntWritable.class); //為job輸出設置value類(lèi) FileInputFormat.addInputPath(job, new Path(otherArgs[0])); //為job設置輸入路徑 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));//為job設置輸出路徑 System.exit(job.waitForCompletion(true) ?0 : 1); //運行job }}
public void map(Object key, Text value, Context context) throws IOException, InterruptedException {…}這里有三個(gè)參數，前面兩個(gè)Object key, Text value就是輸入的key和value，第三個(gè)參數Context context這是可以記錄輸入的key和value，例如：context.write(word, one);此外context還會(huì )記錄map運算的狀態(tài)。
public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {…}reduce函數的輸入也是一個(gè)key/value的形式，不過(guò)它的value是一個(gè)迭代器的形式Iterable<IntWritable> values，也就是說(shuō)reduce的輸入是一個(gè)key對應一組的值的value，reduce也有context和map的context作用一致。
至于計算的邏輯則需要程序員編碼實(shí)現。
首先是：
Configuration conf = new Configuration();
運行MapReduce程序前都要初始化Configuration，該類(lèi)主要是讀取MapReduce系統配置信息，這些信息包括hdfs還有MapReduce，也就是安裝hadoop時(shí)候的配置文件例如：core-site.xml、hdfs-site.xml和mapred-site.xml等等文件里的信息，有些童鞋不理解為啥要這么做，這個(gè)是沒(méi)有深入思考MapReduce計算框架造成，我們程序員開(kāi)發(fā)MapReduce時(shí)候只是在填空，在map函數和reduce函數里編寫(xiě)實(shí)際進(jìn)行的業(yè)務(wù)邏輯，其它的工作都是交給MapReduce框架自己操作的，但是至少我們要告訴它怎么操作啊，比如hdfs在哪里，MapReduce的jobstracker在哪里，而這些信息就在conf包下的配置文件里。
接下來(lái)的代碼是：
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length != 2) { System.err.println("Usage: wordcount <in> <out>"); System.exit(2); }If的語(yǔ)句好理解，就是運行WordCount程序時(shí)候一定是兩個(gè)參數，如果不是就會(huì )報錯退出。至于第一句里的GenericOptionsParser類(lèi)，它是用來(lái)解釋常用hadoop命令，并根據需要為Configuration對象設置相應的值，其實(shí)平時(shí)開(kāi)發(fā)里我們不太常用它，而是讓類(lèi)實(shí)現Tool接口，然后再main函數里使用ToolRunner運行程序，而ToolRunner內部會(huì )調用GenericOptionsParser。
接下來(lái)的代碼是：
Job job = new Job(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class);
第一行就是在構建一個(gè)job，在mapreduce框架里一個(gè)mapreduce任務(wù)也叫mapreduce作業(yè)也叫做一個(gè)mapreduce的job，而具體的map和reduce運算就是task了，這里我們構建一個(gè)job，構建時(shí)候有兩個(gè)參數，一個(gè)是conf這個(gè)就不累述了，一個(gè)是這個(gè)job的名稱(chēng)。
第二行就是裝載程序員編寫(xiě)好的計算程序，例如我們的程序類(lèi)名就是WordCount了。這里我要做下糾正，雖然我們編寫(xiě)mapreduce程序只需要實(shí)現map函數和reduce函數，但是實(shí)際開(kāi)發(fā)我們要實(shí)現三個(gè)類(lèi)，第三個(gè)類(lèi)是為了配置mapreduce如何運行map和reduce函數，準確的說(shuō)就是構建一個(gè)mapreduce能執行的job了，例如WordCount類(lèi)。
第三行和第五行就是裝載map函數和reduce函數實(shí)現類(lèi)了，這里多了個(gè)第四行，這個(gè)是裝載Combiner類(lèi)，這個(gè)類(lèi)和mapreduce運行機制有關(guān)，其實(shí)本例去掉第四行也沒(méi)有關(guān)系，但是使用了第四行理論上運行效率會(huì )更好。
接下來(lái)的代碼：
job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class);
這個(gè)是定義輸出的key/value的類(lèi)型，也就是最終存儲在hdfs上結果文件的key/value的類(lèi)型。
最后的代碼是：
FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1);
第一行就是構建輸入的數據文件，第二行是構建輸出的數據文件，最后一行如果job運行成功了，我們的程序就會(huì )正常退出。
聯(lián)系客服
