MapReduce框架Partitioner分区方法

前言:对于二次排序相信大家也是似懂非懂,我也是一样,对其中的很多方法都不理解诶,所有只有暂时放在一边,当你接触到其他的函数,你知道的越多时你对二次排序的理解也就更深入了,同时建议大家对wordcount的流程好好分析一下,要真正的知道每一步都是干什么的。

1.Partitioner分区类的作用是什么?
2.getPartition()三个参数分别是什么?
3.numReduceTasks指的是设置的Reducer任务数量,默认值是是多少?
扩展:
如果不同类型的数据被分配到了同一个分区,输出的数据是否还是有序的?

在进行MapReduce计算时,有时候需要把最终的输出数据分到不同的文件中,比如按照省份划分的话,需要把同一省份的数据放到一个文件中;按照性别划分的话,需要把同一性别的数据放到一个文件中。我们知道最终的输出数据是来自于Reducer任务。那么,如果要得到多个文件,意味着有同样数量的Reducer任务在运行。Reducer任务的数据来自于Mapper任务,也就说Mapper任务要划分数据,对于不同的数据分配给不同的Reducer任务运行。Mapper任务划分数据的过程就称作Partition。负责实现划分数据的类称作Partitioner。

Partitoner类的源码如下:

package org.apache.hadoop.mapreduce.lib.partition;

import org.apache.hadoop.mapreduce.Partitioner;

/** Partition keys by their {@link Object#hashCode()}. */
public class HashPartitioner<K, V> extends Partitioner<K, V> {

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K key, V value,
                          int numReduceTasks) {
   //默认使用key的hash值与上int的最大值,避免出现数据溢出 的情况
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

}

HashPartitioner是处理Mapper任务输出的,getPartition()方法有三个形参,源码中key、value分别指的是Mapper任务的输出,numReduceTasks指的是设置的Reducer任务数量,默认值是1。那么任何整数与1相除的余数肯定是0。也就是说getPartition(…)方法的返回值总是0。也就是Mapper任务的输出总是送给一个Reducer任务,最终只能输出到一个文件中。 据此分析,如果想要最终输出到多个文件中,在Mapper任务中对数据应该划分到多个区中。那么,我们只需要按照一定的规则让getPartition(…)方法的返回值是0,1,2,3…即可。
大部分情况下,我们都会使用默认的分区函数,但有时我们又有一些,特殊的需求,而需要定制Partition来完成我们的业务,案例如下:
对如下数据,按字符串的长度分区,长度为1的放在一个,2的一个,3的各一个。

河南省;1
河南;2
中国;3
中国人;4
大;1
小;3
中;11

这时候,我们使用默认的分区函数,就不行了,所以需要我们定制自己的Partition,首先分析下,我们需要3个分区输出,所以在设置reduce的个数时,一定要设置为3,其次在partition里,进行分区时,要根据长度具体分区,而不是根据字符串的hash码来分区。核心代码如下:


   public static class PPartition extends Partitioner<Text, Text>{ 
  @Override
  public int getPartition(Text arg0, Text arg1, int arg2) {
    /**     * 自定义分区,实现长度不同的字符串,分到不同的reduce里面     *     * 现在只有3个长度的字符串,所以可以把reduce的个数设置为3     * 有几个分区,就设置为几     * */
   
   String key=arg0.toString();
   if(key.length()==1){
    return 1%arg2;
   }else if(key.length()==2){
    return 2%arg2;
   }else if(key.length()==3){
    return 3%arg2;
   }
   return  0;
  }     
  }

在运行Mapreduce程序时,只需在主函数里加入如下两行代码即可:

job.setPartitionerClass(PPartition.class);
job.setNumReduceTasks(3);//设置为3


    原文作者:MapReduce
    原文地址: https://yq.aliyun.com/articles/413192
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞