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
