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

    }
}