package Mapreduce;

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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

/**
 * 规约：Combiner
 * 
 */
public class CombinerTest {
    public static void main(String[] args) throws Exception {
        //必须要传递的是自定的mapper和reducer的类，输入输出的路径必须指定，输出的类型<k3,v3>必须指定
        //2将自定义的MyMapper和MyReducer组装在一起
        Configuration conf=new Configuration();
        String jobName=CombinerTest.class.getSimpleName();
        //1首先寫job，知道需要conf和jobname在去創建即可
        Job job = Job.getInstance(conf, jobName);
        
        //*13最后，如果要打包运行改程序，则需要调用如下行
        job.setJarByClass(CombinerTest.class);
        
        //3读取HDFS內容：FileInputFormat在mapreduce.lib包下
        FileInputFormat.setInputPaths(job, new Path("hdfs://neusoft-master:9000/data/hellodemo"));
        //4指定解析<k1,v1>的类（谁来解析键值对）
        //*指定解析的类可以省略不写，因为设置解析类默认的就是TextInputFormat.class
        job.setInputFormatClass(TextInputFormat.class);
        //5指定自定义mapper类
        job.setMapperClass(MyMapper.class);
        //6指定map输出的key2的类型和value2的类型  <k2,v2>
        //*下面两步可以省略，当<k3,v3>和<k2,v2>类型一致的时候,<k2,v2>类型可以不指定
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        //7分区(默认1个)，排序，分组，规约 采用 默认
        
        //规约的例子
        job.setCombinerClass(MyReducer.class);
        //接下来采用reduce步骤
        //8指定自定义的reduce类
        job.setReducerClass(MyReducer.class);
        //9指定输出的<k3,v3>类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //10指定输出<K3,V3>的类
        //*下面这一步可以省
        job.setOutputFormatClass(TextOutputFormat.class);
        //11指定输出路径
        FileOutputFormat.setOutputPath(job, new Path("hdfs://neusoft-master:9000/out1"));
        
        //12写的mapreduce程序要交给resource manager运行
        job.waitForCompletion(true);
    }
    private static class MyMapper extends Mapper<LongWritable, Text, Text,LongWritable>{
        Text k2 = new Text();
        LongWritable v2 = new LongWritable();
        @Override
        protected void map(LongWritable key, Text value,//三个参数
                Mapper<LongWritable, Text, Text, LongWritable>.Context context) 
                throws IOException, InterruptedException {
            String line = value.toString();
            String[] splited = line.split("\t");//因为split方法属于string字符的方法，首先应该转化为string类型在使用
            for (String word : splited) {
                //word表示每一行中每个单词
                //对K2和V2赋值
                k2.set(word);
                v2.set(1L);
                context.write(k2, v2);
            }
        }
    }
    private static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
        LongWritable v3 = new LongWritable();
        @Override //k2表示单词，v2s表示不同单词出现的次数，需要对v2s进行迭代
        protected void reduce(Text k2, Iterable<LongWritable> v2s,  //三个参数
                Reducer<Text, LongWritable, Text, LongWritable>.Context context)
                throws IOException, InterruptedException {
            long sum =0;
            for (LongWritable v2 : v2s) {
                //LongWritable本身是hadoop类型，sum是java类型
                //首先将LongWritable转化为字符串，利用get方法
                sum+=v2.get();
            }
            v3.set(sum);
            //将k2,v3写出去
            context.write(k2, v3);
        }
    }
}