package com.cr.skew1_stage_version2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class SkewApp2 {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        //單例作業
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","file:///");
        Job job = Job.getInstance(conf);
        System.setProperty("hadoop.home.dir","E:\\hadoop-2.7.5");

        //設定job的各種屬性
        job.setJobName("SkewApp2");                 //設定job名稱
        job.setJarByClass(SkewApp2.class);              //設定搜尋類
        job.setInputFormatClass(KeyValueTextInputFormat.class);

        //設定輸入路徑
        FileInputFormat.addInputPath(job,new Path(("D:\\skew\\out\\part-r-00000")));
        FileInputFormat.addInputPath(job,new Path(("D:\\skew\\out\\part-r-00001")));
        FileInputFormat.addInputPath(job,new Path(("D:\\skew\\out\\part-r-00002")));
        //設定輸出路徑
        Path path = new Path("D:\\skew\\out2");
        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(path)) {
            fs.delete(path, true);
        }
        FileOutputFormat.setOutputPath(job,path);

        job.setMapperClass(SkewMapper2.class);               //設定mapper類
        job.setReducerClass(SkewReducer1.class);               //設定reduecer類

        job.setMapOutputKeyClass(Text.class);            //設定之map輸出key
        job.setMapOutputValueClass(IntWritable.class);   //設定map輸出value

        job.setOutputKeyClass(Text.class);               //設定mapreduce 輸出key
        job.setOutputValueClass(IntWritable.class);      //設定mapreduce輸出value

        job.setNumReduceTasks(3);
        job.waitForCompletion(true);

    }

}
 

public class KeyValueTextInputFormat extends FileInputFormat<Text, Text> {
    public KeyValueTextInputFormat() {
    }