自定义Partitioner

需求

将统计结果按照手机号，以136、137、138、139开头的数据分别放到一个独立的文件中，其他开头的放到一个文件中。（分区）

输入数据

1863157985066   120.196.100.82 	2481    24681    200
1363157995033   120.197.40.4      264    0    200
1373157993055   120.196.100.99    132    1512    200
1393154400022   120.197.40.4      240    0    200
1363157993044   120.196.100.99    1527    2106    200
1397157993055   120.197.40.4      4116    1432    200
1463157993055   120.196.100.99    1116    954    200
1383157995033   120.197.40.4      3156    2936    200
1363157983019   120.196.100.82    240    0    200
1383154400022   120.197.40.4      6960    690    200
1363157973098   120.197.40.4      3659    3538    200
1373157993055   120.196.100.99    1938    180    200
1363154400022   120.196.100.99    918    4938    200
1393157993055   120.197.40.4      180    180    200
1363157984040   120.197.40.4      1938    2910    200
.......

具体实现：

第一步：自定义Mapper：

public class PhoneMapper extends Mapper {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();   //拿到一行数据
        String[] fields =  line.split("\\s+");  //切分成各个字段
        String phoneNumber = fields[0]; //拿到手机号的字段
        //封装数据为key-value进行输出
        context.write(new Text(phoneNumber), value);
    }
}

第二步：自定义Partitioner

public class PhonePartitioner extends Partitioner {
    @Override
    public int getPartition(Text key, Text value, int numPartitions) {
        String preNum = key.toString().substring(0, 3); // 1 获取电话号码的前三位
        int partition = 4;
        switch (preNum) {
            case "136":
                partition = 0;
                break;
            case "137":
                partition = 1;
                break;
            case "138":
                partition = 2;
                break;
            case "139":
                partition = 3;
                break;
            default:
                break;
        }
        return partition;
    }
}

第三步：自定义Reducer

public class PhoneReducer extends Reducer {
    int index = 0;
    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        index++;
        context.write(new LongWritable(index), values.iterator().next());
    }
}

第四步：自定义Driver

public class PhoneDriver {
    public static void main(String[] args) throws Exception {
        args = new String[2];
        args[0] = "src/main/resources/phonei";
        args[1] = "src/main/resources/phoneo";

        // 1 获取配置信息，或者job对象实例
        Configuration cfg = new Configuration();
        //设置本地模式运行（即使项目类路径下core-site.xml文件，依然采用本地模式）
        cfg.set("mapreduce.framework.name", "local");
        cfg.set("fs.defaultFS", "file:///");
        Job job = Job.getInstance(cfg);

        // 2 指定本程序的jar包所在的本地路径
        job.setJarByClass(PhoneDriver.class);

        // 3 指定本业务job要使用的mapper/Reducer业务类
        job.setMapperClass(PhoneMapper.class);
        job.setReducerClass(PhoneReducer.class);

        // 4 指定mapper输出数据的kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        // 5 指定最终输出的数据的kv类型
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(Text.class);

        // 8 指定自定义数据分区
        job.setPartitionerClass(PhonePartitioner.class);
        // 9 同时指定相应数量的reduce task(必须指定)
        job.setNumReduceTasks(5);  //----①

        // 6 指定job的输入原始文件所在目录
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

本示例自定义分区数为4，依次调整①的代码：

• job.setNumReduceTasks(1); 会正常运行，不过只会产生一个输出文件
• job.setNumReduceTasks(2); 会报错
• job.setNumReduceTasks(6); 大于4，程序会正常运行，会产生空文件