package cn.hdfs.mr.example;

import java.io.IOException;

import java.util.Random;

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.slf4j.Logger;

import org.slf4j.LoggerFactory;

import cn.hdfs.utils.ConfigUtils;

/**

 *

 * @author dengjie

 * @date 2015年03月13日

 * @description Wordcount的例子是一个比较经典的mapreduce例子，可以叫做Hadoop版的hello world。

 *              它将文件中的单词分割取出，然后shuffle，sort（map过程），接着进入到汇总统计

 *              （reduce过程），最后写道hdfs中。基本流程就是这样。

 */

public class WordCount {

    private static Logger log = LoggerFactory.getLogger(WordCount.class);

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

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

    private Text word = new Text();

    /*

     * 源文件：a b b

     *

     * map之后：

     *

     * a 1

     *

     * b 1

     *

     * b 1

     */

    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());// 按空格分割单词

        context.write(word, one);// 每次统计出来的单词+1

        }

    }

    }

    /*

     * reduce之前：

     *

     * a 1

     *

     * b 1

     *

     * b 1

     *

     * reduce之后:

     *

     * a 1

     *

     * b 2

     */

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

    }

    }

    @SuppressWarnings("deprecation")

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

    Configuration conf1 = new Configuration();

    Configuration conf2 = new Configuration();

    long random1 = new Random().nextLong();// 重定下输出目录1

    long random2 = new Random().nextLong();// 重定下输出目录2

    log.info("random1 -> " + random1 + ",random2 -> " + random2);

    Job job1 = new Job(conf1, "word count1");

    job1.setJarByClass(WordCount.class);

    job1.setMapperClass(TokenizerMapper.class);// 指定Map计算的类

    job1.setCombinerClass(IntSumReducer.class);// 合并的类

    job1.setReducerClass(IntSumReducer.class);// Reduce的类

    job1.setOutputKeyClass(Text.class);// 输出Key类型

    job1.setOutputValueClass(IntWritable.class);// 输出值类型  

    Job job2 = new Job(conf2, "word count2");

    job2.setJarByClass(WordCount.class);

    job2.setMapperClass(TokenizerMapper.class);

    job2.setCombinerClass(IntSumReducer.class);

    job2.setReducerClass(IntSumReducer.class);

    job2.setOutputKeyClass(Text.class);

    job2.setOutputValueClass(IntWritable.class);

    // FileInputFormat.addInputPath(job, new

    // Path(String.format(ConfigUtils.HDFS.WORDCOUNT_IN, "test.txt")));

    // 指定输入路径

    FileInputFormat.addInputPath(job1, new Path(String.format(ConfigUtils.HDFS.WORDCOUNT_IN, "word")));

    // 指定输出路径

    FileOutputFormat.setOutputPath(job1, new Path(String.format(ConfigUtils.HDFS.WORDCOUNT_OUT, random1)));

    FileInputFormat.addInputPath(job2, new Path(String.format(ConfigUtils.HDFS.WORDCOUNT_IN, "word")));

    FileOutputFormat.setOutputPath(job2, new Path(String.format(ConfigUtils.HDFS.WORDCOUNT_OUT, random2)));

    boolean flag1 = job1.waitForCompletion(true);// 执行完MR任务后退出应用

    boolean flag2 = job1.waitForCompletion(true);

    if (flag1 && flag2) {

        System.exit(0);

    } else {

        System.exit(1);

    }

    }

}