MapReduce实现自定义分区与排序

ゝ一世哀愁。 2023-10-11 21:30 43阅读 0赞

##### 分区数据排序 #####

Hadoop中默认的numReduceTask数量为1，也就是说所有数据将来都会被输出为一个分区。如果想根据自定义的业务逻辑实现分区，则需要继承Partitioner类。

@Public
    @Stable
    public abstract class Partitioner<KEY, VALUE> {
        
        public Partitioner() {
        
        }
    
        public abstract int getPartition(KEY var1, VALUE var2, int var3);
    }

这个类的泛型非常重要，<KEY,VALUE>分别对应的Map输出的KEY,VALUE，那map输出的k,v就完全对应reduce的输入,所以这个接口的逻辑就完全实现了最终的数据有多少个分区。

[
    {
        name : "127"},
    {
        name : "127"},
    {
        name : "128"},
    {
        name : "128"},
    {
        name : "129"},
    {
        name : "129"},
    {
        name : "126"},
    {
        name : "126"},
    {
        name : "125"},
    {
        name : "125"},
    ]

如果以name为key实现分区，name相同的数据放到一个分区，这组数据将来会有五个分区，此时就需要继承Partitioner类，实现getPartition方法。

public class NamePartitioner extends Partitioner<Text,Text> {
        
    
        @Override
        public int getPartition(Text text, Text text2, int i) {
        
            String string = text.toString();
            if (string.equals("125")){
        
                return 0;
            }else if (string.equals("126")){
        
                return 1;
            }else if (string.equals("127")){
        
                return 2;
            }else if (string.equals("128")){
        
                return 3;
            }else if (string.equals("129")){
        
                return 4;
            }
            return -1;
        }
    }

定义完分区类以后，需要在驱动Job类中挂载这个类，并指定NumReduceTask个数。

job.setNumReduceTasks(5);
            //设置分区策略
            job.setPartitionerClass(NamePartitioner .class);

在此案例中，如果运行此程序需要启动一个ApplicationMaster，1个MapTask，5个ReduceTask，总共7个Container，1个Container分配1G最小内存，1个CPU的话，则运行此任务需要7G内存 + 7个CPU。

*  设置分区的注意事项

1.  job.setNumReduceTasks() 的参数一定要保持和分区器的分区数一致，例如getPartitioner方法返回了6个分区的情况，而job.setNumReduceTasks(5);只设置了5个分区，此时有一个分区的数就不知道往哪里放了，但是此时如果是设置了多个分区，但是getPartiton方法只返回了一种分区情况，此时也不会报错，所有的mapTask都会输出到同一个分区。
2.  job.setPartitionerClass(NamePartitioner .class); //一定要将自己的分区实现类挂载上
3.  如果自定义的Bean要是作为key的话，必须实现Writable与CompareWritale接口。

##### 排序接口 #####

如果自定义的Bean要作为MapTask的key，则必须要实现CompareWritale接口。此接口要实现的逻辑和集合的Sort方法是一致的。

public class FlowBean implements Writable, WritableComparable<FlowBean> {
        
    
        @Override
        public int compareTo(FlowBean o) {
        
            return (int) (this.upFlow - o.upFlow);
        }
    }

写过冒泡排序算法的都知道，冒泡排序是亮亮比较，如果想升序，则一定是比较的两个对象 前者 - 后者 > 0，这样才会交换两个对象的位置。  
如果是降序，则后者 - 前者 > 0，这样交换两者的位置后就变成了后者在前面。