mapreduce.input.fileinputformat.split.minsize=1 默認(rèn)值為1
mapreduce.input.fileinputformat.split.maxsize=Long.MAXValue 默認(rèn)值Long.MAXValue

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {
    @Override
    public int getPartition(Text text, FlowBean flowBean, int i) {
        // 獲取手機(jī)號前三位prePhone
        String phone = text.toString();
        String prePhone = phone.substring(0, 3);
        //定義一個分區(qū)號變量partition,根據(jù)prePhone 設(shè)置分區(qū)號
        int partition;
        if ("136".equals(prePhone)) {
            partition = 0;
        } else if ("137".equals(prePhone)) {
            partition = 1;
        } else if ("138".equals(prePhone)) {
            partition = 2;
        } else if ("139".equals(prePhone)) {
            partition = 3;
        } else {
            partition = 4;
        }
        //最后返回分區(qū)號partition
        return partition;
    }
}

public class FlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 1.獲取job對象
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        // 2.關(guān)聯(lián)本Driver類
        job.setJarByClass(FlowDriver.class);
        // 3.關(guān)聯(lián)Mapper和Reducer類
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);
        // 4.設(shè)置Map端輸出KV類型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
        // 5.設(shè)置最終輸出KV類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
        // 6.設(shè)置程序的輸入和輸出路徑
        FileInputFormat.setInputPaths(job, new Path("C:\\install\\temp\\input\\input02\\phone_data.txt"));
        FileOutputFormat.setOutputPath(job, new Path("C:\\install\\temp\\output\\output06"));
        // 8.指定自定義分區(qū)器
        job.setPartitionerClass(ProvincePartitioner.class);
        // 9.同時也指定相應(yīng)數(shù)量的ReduceTask--對應(yīng)的參數(shù)mapreduce.job.reduces，默認(rèn)為1
        job.setNumReduceTasks(5);
        // 7.提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

public class WordCountCombiner extends Reducer<Text, IntWritable, Text,IntWritable> {
    private IntWritable outV = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        outV.set(sum);
        context.write(key,outV);
    }
}

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 一獲取二關(guān)聯(lián)三設(shè)置一提交
        // 1.獲取配置信息及Job對象
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        // 2.關(guān)聯(lián)本Driver程序的類
        job.setJarByClass(WordCountDriver.class);
        // 3.關(guān)聯(lián)Mapper和Reducer的業(yè)務(wù)類
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        // 4.設(shè)置Mapper輸出的KV類型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        // 5.設(shè)置最終輸出的KV類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 6.設(shè)置輸入和輸出路徑
        //FileInputFormat.setInputPaths(job, new Path(args[0]));
        //FileOutputFormat.setOutputPath(job, new Path(args[1]));
        FileInputFormat.setInputPaths(job, new Path("C:\\install\\temp\\input\\hadoop.txt"));
        FileOutputFormat.setOutputPath(job, new Path("C:\\install\\temp\\output\\output01-2"));
        // 8.設(shè)置Combiner類--方式一
        //job.setCombinerClass(WordCountCombiner.class);
        // 方式二：其新建的WordCountCombiner的reduce方法處理與正常的WordCountReducer中的reduce方法處理邏輯是一樣
        // 因此可以直接用此類作為combiner類
        job.setCombinerClass(WordCountReducer.class);
        // 9.設(shè)置ReduceTasks的數(shù)量--這樣就沒有reduce階段，就不會有shuffle，Combiner也就沒有用，直接由map輸出，
        // 文件名為part-m-00000，就是不part-r-00000，兩者結(jié)果是不一樣的
        // 即如果沒有reduce階段，即使設(shè)置了combiner也不起作用
        // job.setNumReduceTasks(0);
        // 7.提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1); // 0-正常退出 非0（1）異常終止（結(jié)束）
    }
}