亲宝软件园·资讯

展开

hadoop 自定义分区

小码农叔叔 人气:0

分区概念

分区这个词对很多同学来说并不陌生,比如Java很多中间件中,像kafka的分区,mysql的分区表等,分区存在的意义在于将数据按照业务规则进行合理的划分,方便后续对各个分区数据高效处理

Hadoop分区

hadoop中的分区,是把不同数据输出到不同reduceTask ,最终到输出不同文件中

hadoop 默认分区规则

以下是Partition 类中摘取出来的源码,还是很容易懂的

hash分区代码演示

下面是wordcount案例中的driver部分的代码,默认情况下我们不做任何设置,最终输出一个统计单词个数的txt文件,如果我们在这段代码中添加这样一行

再次运行下面的程序后,会出现什么结果呢?

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);
    }

}

可以看到,最终输出了2个统计结果文件,每个文件中的内容有所不同,这就是默认情况下,当reducer个数设置为多个时,会按照hash分区算法计算结果并输出到不同分区对应的文件中去

自定义分区步骤

业务需求

将下面文件中 的人物名称按照姓氏,“马”姓的放入第一个分区,“李”姓的放入第二个分区,其他的放到其他第三个分区中

自定义分区

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;
    }
}

将自定义分区关联到Driver类中,注意这里的ReduceTasks个数和自定义的分区数量保持一致

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

下面运行Driver类,观察最终的输出结果,也是按照预期,将不同的姓氏数据输出到了不同的文件中

关于自定义分区的总结

加载全部内容

相关教程
猜你喜欢
用户评论