package com.my.hadoop.mapreduce.wordcount;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;

/**
 * MapReduce中的WordCount
 * @author yao
 *
 */
public class WordCount {

/**
     * MapReduce中的map函数的泛型
     * KEYIN        map函数读取文件行内容的偏移量为key
     * VALUEIN         map函数读取文件行内容
     * KEYOUT        map函数处理后输出到reduce函数的key
     * VALUEOUT        map函数处理后输出到reduce函数的value
     * @author yao
     *
     */
    static class WcMap extends Mapper<LongWritable, Text, Text, LongWritable>{
        private static final LongWritable ONE = new LongWritable(1l);
        private Text word = new Text();
        public void map(LongWritable key, Text value, Context context) throws IOException ,InterruptedException {
            String[] words = value.toString().split(" ");
            for (String w : words) {
                word.set(w);
                context.write(word, ONE);
            }
        }
    }
    
    /**
     * MapReduce中的reduce函数的泛型
     * KEYIN        reduce函数读取map函数输出的key
     * VALUEIN        reduce函数读取map函数输出的value
     * KEYOUT        reduce函数处理后输出到hdfs上文件的key
     * VALUEOUT        reduce函数处理后输出到hdfs上文件的value
     * @author yao
     *
     */
    static class WcReduce extends Reducer<Text, LongWritable, Text, LongWritable>{
        public void reduce(Text key, Iterable<LongWritable> value, Context context) throws java.io.IOException ,InterruptedException {
            long count = 0;
            for (LongWritable i : value) {
                count += i.get();
            }
            context.write(key, new LongWritable(count));
        }
    }
    
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();                                            //new配置对象，默认读取顺序是default-site.xml<core-site.xml
        
        String[] paths = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (paths.length != 2) {
            System.err.println("Usage: " + WordCount.class.getName() + " <in> <out>");
            System.exit(2);
        }
        
        Job job = Job.getInstance(conf, WordCount.class.getSimpleName());                    //1.x是new Job，2.x为Job.getInstance
        job.setJarByClass(WordCount.class);                                                    //设置main方法所在的类
        
        FileInputFormat.setInputPaths(job, new Path(args[0]));                                //设置当前作业的输入路径（可有多个输入路径）
        job.setMapperClass(WcMap.class);                                                    //指定自定义的map函数
        job.setMapOutputKeyClass(Text.class);                                                //指定自定义map函数的输出到reduce函数的key类型
        job.setMapOutputValueClass(LongWritable.class);                                        //指定自定义map函数的输出到reduce函数的value类型
        
        job.setCombinerClass(WcReduce.class);                                                //在map函数输出到reduce函数进行本地合并以减少网络传输的带宽资源（根据需求使用，并不适用所有业务）
        
        job.setReducerClass(WcReduce.class);                                                //指定自定义的reduce函数
        job.setOutputKeyClass(Text.class);                                                    //指定自定义的reduce函数输出到hdfs的key类型
        job.setOutputValueClass(LongWritable.class);                                        //指定自定义的reduce函数输出到hdfs的value类型
        FileOutputFormat.setOutputPath(job, new Path(args[1]));                                //设置当前作业的输出到hdfs的路径（只有一个输出路径且该路径必须不存在）
        
        int status = job.waitForCompletion(true) ? 0 : 1;                                    //提交作业：true是打印作业进度详情，false则是不打印
        System.exit(status);
    }

}