Hadoop 初学（一）wordcount

     Hadoop的中文文档更新速度有点慢，照着敲的大部分API都是过时的，而且设置也有些问题，按照官网的方法，跑不起来，报路径找不到。


简单敲了一下代码先熟练一下基本的编程模式和API
官网的例子是统计单词，在一个目录下有多个文本文件，要统计其中不同单词出现的次数。

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;




public class WordCount {
    public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable>{
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        @Override
        public void map(LongWritable key, Text value,
                OutputCollector<Text, IntWritable> output, Reporter reporter)
                throws IOException {
            String line = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(line);
            while(tokenizer.hasMoreTokens()){
                word.set(tokenizer.nextToken());
                output.collect(word, one);
            }

        }




    }

    public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable>{

        @Override
        public void reduce(Text key, Iterator<IntWritable> values,
                OutputCollector<Text, IntWritable> output, Reporter reporter)
                throws IOException {
            int sum = 0;
            while(values.hasNext()){
                sum+=values.next().get();
            }
            output.collect(key, new IntWritable(sum));

        }

    }

    /**
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        JobConf conf = new JobConf(WordCount.class);
        conf.setJobName("word count");

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

        conf.setMapperClass(Map.class);
        conf.setCombinerClass(Reduce.class);
        conf.setReducerClass(Reduce.class);

        conf.setInputFormat(TextInputFormat.class);
        conf.setOutputFormat(TextOutputFormat.class);

        FileInputFormat.setInputPaths(conf, new Path(args[0]));
        FileOutputFormat.setOutputPath(conf,new Path(args[1]));

        JobClient.runJob(conf);

    }

}

上面是官网的代码，代码基本没问题，这个程序的流程是：
1、准备一个Mapper
     Mapper的作用是，读入一行文本，将文本分隔成单词串，将每个单词的数目设为1，并分发下去
2、准备一个Reducer
    Reducer的工作恰好相反，它是收集同一个单词的所有结果，并进行统计
3、准备一个Combiner
     这里的Combiner和Reducer作用相同
3、配置好一个Job
     设置Job的key,value类型，设置好Mapper,Reducer,Combiner，输入输出格式以及输入输出路径
4、运行Job
      直接运行

好了，下面设置运行参数，代码里又两个args参数要设置，分别是输入输出，输入是需要文本文件的，所以还要准备好输入文件：
1、首先在本地准备两个文本文件file01,file02
file01

Hello World Goodbye World

file 02

Hello Hadoop Goodbye Hadoop

2、在hadoop节点上建立输入文件路径

./bin/hadoop fs -mkdir /usr/kenvi/wordcount/input

3、将文本文件传到节点服务器上去

./bin/hadoop fs -put file01 /usr/kenvi/wordcount/input/file01
./bin/hadoop fs -put file01 /usr/kenvi/wordcount/input/file01

这样，服务器上的环境就部署好了，还差最后一部，设置本地的参数，官方文档是通过以下命令来运行的：

$ bin/hadoop jar /usr/joe/wordcount.jar org.myorg.WordCount /usr/joe/wordcount/input /usr/joe/wordcount/output

指定的输入输出路径分别是/usr/joe/wordcount/input 和/usr/joe/wordcount/output，这样直接跑是运行不起来的，会报路径找不到。应该分别设置成hdfs://host:9000/usr/kenvi/wordcount/input   和hdfs:///host:9000/usr/kenvi/wordcount/output
其中host为hdfs服务器的名字或者ip

这样就可以正常运行了，运行完后，可以在服务器上的输出目录下看到一个文件part-0000，用

./bin/hadoop fs -cat /usr/kenvi/wordcount/output/part-00000

可以查看文件内容：

Goodbye	1
Hadoop	2
Hello	1