WordCount源码详解

1.源码解释

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;
 
public class WordCount {
 
  public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{

 

 /*定义hadoop数据类型IntWritable实例one，并且赋值为1,为什么是1呢?

因为不管一个单词出现几次,我们都会直接输出1

 假如输入数据是 nice to nice 那么用context.write(word,one)处理的结果就是

nice   1

to       1

nice   1

假如赋值是2 那么输出就变成了,

nice  2

to      2

nice   2

*/

 

    private final static IntWritable one = new IntWritable(1);
 
 //定义hadoop数据类型Text
    private Text word = new Text();
 
    //实现map函数,其中countext参数可以记录key和value
    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());  //itr.nextToken获取单词,word.set() 方法将Java数据类型转换成hadoop数据类型转换,只有这样才能输出
        context.write(word, one);   //按照格式输出结果,比如nice 1  

      }
    }
  }
 
  public static class IntSumReducer    //继承泛型类reducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();
 
    public void reduce(Text key, Iterable<IntWritable> values, Context context

) throws IOException,  InterruptedException {

   //用sum来记录相同单词的数量,是java类型的数据
      int sum = 0;
      for (IntWritable val : values) {      //遍历得到结果
        sum += val.get();        //因为val的数据类型是IntWritable,所以需要将IntWritable类型的数据转换成java可以识别的数据类型,Intwritable.set(),   IntWritable.get()都能进行数据转换,set()方法是将Java数据类型转换成hadoop可识别的数据类型,而get()方法是将hadoop的数据类型转换成java可识别的数据类型
      }

      result.set(sum);     //将java的数据类型转换成Hadoop可识别的数据类型
      context.write(key, result);   //输出结果到hdfs
    }
  }
 
  public static void main(String[] args) throws Exception {
  //读取hadoop的配置参数,也就是安装hadoop时候的配置文件例如：core-site.xml、hdfs-site.xml和mapred-site.xml等等文件里的信息
    Configuration conf = new Configuration();   
 
//GenericOptionsParser类，它是用来解释常用hadoop命令，并根据需要为Configuration对象设置相应的值
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
 
//运行WordCount程序时候一定是两个参数，如果不是就会报错退出,其实这2个参数的作用就是告诉机器输入文件的地址和输出文件的地址
    if (otherArgs.length != 2) {      
      System.err.println("Usage: wordcount <in> <out>");
      System.exit(2);
    }
 
    Job job = new Job(conf, "word count");  //创建一个任务,第二个参数word count是任务的名称
    job.setJarByClass(WordCount.class);   
 
    job.setMapperClass(TokenizerMapper.class); //装载程序员编写好的计算程序，例如我们的程序类名就是WordCount了。这里我要做下纠正，虽然我们编写mapreduce程序只需要实现map函数和reduce函数，但是实际开发我们要实现三个类，第三个类是为了配置mapreduce如何运行map和reduce函数，准确的说就是构建一个mapreduce能执行的job了，例如WordCount类。
 
 
    job.setCombinerClass(IntSumReducer.class); 
    job.setReducerClass(IntSumReducer.class);  //指定要使用的reduce类
    job.setOutputKeyClass(Text.class);  
    job.setOutputValueClass(IntWritable.class); //定义输出结果的key/value的类型，也就是最终存储在hdfs上结果文件的key/value的类型。
 
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));   //确定输入文件的路径
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));  //确定输出文件的路径
 
    System.exit(job.waitForCompletion(true) ? 0 : 1); //如果job成功运行,我们的程序就会正常退出
  }
}

 

2.自己动手写出WordCount程序

分析WordCount的示例程序,可以看出一个基本的结构

底层的东西,hadoop的框架已经帮我们搞定了,我们剩下的工作其实就是设计以下3个方面

• mapper类
• reducer类
• main() 方法

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;
 
public class xxx {
 
  public static class MyMapper extends Mapper<xxx, xxx, xxx, xxx>{
 
    public void map(xxx key, xxx value, Context context) throws IOException, InterruptedException {
    编写你的代码

      }
 }
 
  public static class MyReducer extends Reducer<xxx,xxx,xxx,xxx> {

    public void reduce(xxx key, xxx values, Context context) throws IOException, InterruptedException {
    编写你的代码  

    }
 }
 
  public static void main(String[] args) throws Exception {
    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, "xxx");
    job.setJarByClass(xxx.class);   //根据自己编写的类来填写
    job.setMapperClass(xxx.class);
    job.setCombinerClass(xxx.class);
    job.setReducerClass(xxx.class);
    job.setOutputKeyClass(xxx.class);
    job.setOutputValueClass(xxx.class);
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }