Hadoop WordCount详解（二）

Hadoop集群WordCount详解（二）

• 源代码程序

• 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.examples.WordCount.TokenizerMapper;
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代码
 * @author lzxyzq
 *
 */

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();

        public void map(Object key,Text value,Context context)throws IOException,InterruptedException{

            StringTokenizer st = new StringTokenizer(value.toString());
                while(st.hasMoreTokens()){
                    word.set(st.nextToken());
                    context.write(word, one);
                }
        }
    }

    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();
            }
        result.set(sum);
        context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception{

        Configuration config = new Configuration();
        String[] otherArgs = new GenericOptionsParser(config,args).getRemainingArgs();

        if(otherArgs.length!=2){
            System.out.println("Usage:workcount<in><out>");
            System.exit(2);
        }

        Job job = Job.getInstance(config, "word count2");

        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }

        FileOutputFormat.setOutputPath(job,new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}


/**
 * API
 * java.util.StringTokenizer 
 * The following is one example of the use of the tokenizer. The code:

        StringTokenizer st = new StringTokenizer("this is a test");
        while (st.hasMoreTokens()) {
        System.out.println(st.nextToken());
        }
        prints the following output:
        this
        is
        a
        test
 */

• 1）Map过程

public static class Tokenizermapper extends Mapper<Object,Text,Text,IntWritable>{

        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key,Text value,Context context)throws IOException,InterruptedException{

            StringTokenizer st = new StringTokenizer(value.toString());
            while(st.hasMoreTokens()){
                word.set(st.nextToken());
                context.write(word, one);
            }
        }
    }

Map过程需要继承org.apache.hadoop.mapreduce包中Mapper类，并重写其map方法。通过在map方法中添加两句把key值和value值输出到控制台的代码，可以发现map方法中value值存储的是文本文件中的一行（以回车符为行结束标记），而key值为该行的首字母相对于文本文件的首地址的偏移量。然后StringTokenizer类将每一行拆分成为一个个的单词，并将(word,1)作为map方法的结果输出，其余的工作都交有MapReduce框架处理。

• 2) Reduce过程

public void 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);
        }
    }

Reduce过程需要继承org.apache.hadoop.mapreduce包中Reducer类，并重写其reduce方法。Map过程输出(key,values)中key为单个单词，而values是对应单词的计数值所组成的列表，Map的输出就是Reduce的输入，所以reduce方法只要遍历values并求和，即可得到某个单词的总次数。

• 3）执行MapReduce任务

public static void main(String[] args) throws Exception{

        Configuration config = new Configuration();
        String[] otherArgs = new GenericOptionsParser(config,args).getRemainingArgs();

        if(otherArgs.length!=2){
            System.out.println("Usage:workcount<in><out>");
            System.exit(2);
        }

        Job job = Job.getInstance(config, "word count2");
        //main函数调用Jobconf类来对MapReduce Job进行初始化，然后调用setJobName()方法命名这个Job。
        //对Job进行合理的命名有助于更快地找到Job，以便在JobTracker和Tasktracker的页面中对其进行监视。
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
         /**
         *然后设置Job处理的Map（拆分）、Combiner（中间结果合并）以及Reduce（合并）的相关处理类。
         *这里用Reduce类来进行Map产生的中间结果合并，避免给网络数据传输产生压力。
         */
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
         /**
         * 接着设置Job输出结果<key,value>的中key和value数据类型，因为结果是<单词,个数>，
         * 所以key设置为"Text"类型，相当于Java中String类型。
         * Value设置为"IntWritable"，相当于Java中的int类型。
         */
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }

        FileOutputFormat.setOutputPath(job,new Path(otherArgs[otherArgs.length - 1]));
         /**
         * 接着就是调用setInputPath()和setOutputPath()设置输入输出路径。
         */
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

在MapReduce中，由Job对象负责管理和运行一个计算任务，并通过Job的一些方法对任务的参数进行相关的设置。此处设置了使用TokenizerMapper完成Map过程中的处理和使用IntSumReducer完成Combine和Reduce过程中的处理。还设置了Map过程和Reduce过程的输出类型：key的类型为Text，value的类型为IntWritable。任务的输出和输入路径则由命令行参数指定，并由FileInputFormat和FileOutputFormat分别设定。完成相应任务的参数设定后，即可调用job.waitForCompletion()方法执行任务。

2、WordCount处理过程

• 1）将文件拆分成splits，由于测试用的文件较小，所以每个文件为一个split，并将文件按行分割形成(key,value)对，如图1-1所示。这一步由MapReduce框架自动完成，其中偏移量（即key值）包括了回车所占的字符数（Windows和Linux环境会不同）。

• 2）将分割好的(key,value)对交给用户定义的map方法进行处理，生成新的(key,value)对，如图1-2所示。

• 3）得到map方法输出的(key,value)对后，Mapper会将它们按照key值进行排序，并执行Combine过程，将key至相同value值累加，得到Mapper的最终输出结果。如图1-3所示。

• 4）Reducer先对从Mapper接收的数据进行排序，再交由用户自定义的reduce方法进行处理，得到新的(key,value)对，并作为WordCount的输出结果，如图1-4所示。

3、额外补充

（1）InputFormat和InputSplit

　　InputSplit是Hadoop定义的用来传送给每个单独的map的数据，InputSplit存储的并非数据本身，而是一个分片长度和一个记录数据位置的数组。生成InputSplit的方法可以通过InputFormat()来设置。

　　当数据传送给map时，map会将输入分片传送到InputFormat，InputFormat则调用方法getRecordReader()生成RecordReader，RecordReader再通过creatKey()、creatValue()方法创建可供map处理的（key,value）对。简而言之，InputFormat()方法是用来生成可供map处理的（key,value）对的。

　　Hadoop预定义了多种方法将不同类型的输入数据转化为map能够处理的（key,value）对，它们都继承自InputFormat，分别是：

InputFormat

    |---BaileyBorweinPlouffe.BbpInputFormat

    |---ComposableInputFormat

    |---CompositeInputFormat

    |---DBInputFormat

    |---DistSum.Machine.AbstractInputFormat

    |---FileInputFormat

        |---CombineFileInputFormat

        |---KeyValueTextInputFormat

        |---NLineInputFormat

        |---SequenceFileInputFormat

        |---TeraInputFormat

        |---TextInputFormat

其中TextInputFormat是Hadoop默认的输入方法，在TextInputFormat中，每个文件（或其一部分）都会单独地作为map的输入，而这个是继承自FileInputFormat的。之后，每行数据都会生成一条记录，每条记录则表示成(key,value)形式：

• key值是每个数据的记录在数据分片中字节偏移量，数据类型是LongWritable；　　
• value值是每行的内容，数据类型是Text。

（2）OutputFormat

　　每一种输入格式都有一种输出格式与其对应。默认的输出格式是TextOutputFormat，这种输出方式与输入类似，会将每条记录以一行的形式存入文本文件。不过，它的键和值可以是任意形式的，因为程序内容会调用toString()方法将键和值转换为String类型再输出。

（3）Map类中map方法分析

　Map类继承自MapReduceBase，并且它实现了Mapper接口，此接口是一个规范类型，它有4种形式的参数，分别用来指定map的输入key值类型、输入value值类型、输出key值类型和输出value值类型。在本例中，因为使用的是TextInputFormat，它的输出key值是LongWritable类型，输出value值是Text类型，所以map的输入类型为(LongWritable,Text)。在本例中需要输出(word,1)这样的形式，因此输出的key值类型是Text，输出的value值类型是IntWritable。

　　实现此接口类还需要实现map方法，map方法会具体负责对输入进行操作，在本例中，map方法对输入的行以空格为单位进行切分，然后使用OutputCollect收集输出的(word,1)。

（4）Reduce类中reduce方法分析

　Reduce类也是继承自MapReduceBase的，需要实现Reducer接口。Reduce类以map的输出作为输入，因此Reduce的输入类型是(Text，Intwritable)。而Reduce的输出是单词和它的数目，因此，它的输出类型是(Text,IntWritable)。Reduce类也要实现reduce方法，在此方法中，reduce函数将输入的key值作为输出的key值，然后将获得多个value值加起来，作为输出的值。