hadoop全面解读自定义分区

public class DemoJobDriver {

    public static void main(String[] args) throws Exception {

        //1、获取job
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        //2、设置jar路径
        job.setJarByClass(DemoJobDriver.class);

        //3、关联mapper 和 Reducer
        job.setMapperClass(DemoMapper.class);
        job.setReducerClass(DemoReducer.class);

        //4、设置 map输出的 key/val 的类型
        job.setMapOutpuTKEyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5、设置最终输出的key / val 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //6、设置最终的输出路径
        String inputPath = "F:\\网盘\\csv\\hello.txt";
        String outPath = "F:\\网盘\\csv\\wordcount\\hello_result.txt";

        //设置输出文件为2个
        job.setNumReduceTasks(2);

        FileInputFORMat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outPath));

        // 7 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }

}

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapReduce.Partitioner;
import org.apache.hadoop.io.Text;

public class MyPartioner extends Partitioner<Text, IntWritable> {

    @Override
    public int getPartition(Text text, IntWritable intWritable, int partion) {
        String key = text.toString();
        if(StringUtils.isNotEmpty(key.trim())){
            if(key.startsWith("马")){
                partion = 0;
            }else if(key.startsWith("李")){
                partion = 1;
            }else {
                partion = 2;
            }
        }
        return partion;
    }
}

job.setNumReduceTasks(3);
job.setPartitionerClass(MyPartioner.class);