当前位置：首页 > news >正文

Hadoop3教程（十三）：MapReduce中的分区

news 2025/6/26 15:20:29

文章目录

（96）默认HashPartitioner分区
（97）自定义分区案例
（98）分区数与Reduce个数的总结
参考文献

（96）默认HashPartitioner分区

分区，是Shuffle里核心的一环，不同分区的数据最终会被送进不同的ReduceTask去处理。之前的几个小节里也都讲过分区。
Hadoop里默认的分区方式是HashPartitioner分区，核心代码：

public class HashPartitioner<K, V> extends Partitioner<K, V> {public int getPartition(K key, V value, iint numReduceTasks) {return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;}
}

在HashPartitioner里，每个key分到哪个ReduceTask（可以理解成Key属于哪个分区），是根据每个key的hashCode对ReduceTask的个数取模得到的，用户是没法控制的。

这里是为什么还要& Integer.MAX_VALUE呢？

主要是为了防止溢写，通过& Integer.MAX_VALUE，将key的hash值控制在Integer.MAX_VALUE及之下。

从代码里看，在往环形缓冲区写的时候，如果识别到numReduceTasks > 1，则启用HashPartitioner分区，如果numReduceTasks = 1，那就不启用了，直接return numReduceTasks - 1。

我们也可以自定义Partitioner，自定义类需要继承Partitioner类，并重写里面的getPartition()方法。

public class CustomPartitioner extendsPartitioner<Text, FlowBean>{@overridepublic int getPartition(Text key, FlowBean value, int numPartitions){//控制分区代码逻辑。。。。。。return partition;}}

然后在驱动类里，设置上写好的自定义Partitioner：

job.setPartitionerClass(CustomPartitioner.class);

最后再设置上ReduceTask的数量：

job.setNumReduceTasks(5);

如果不设置ReduceTask的数量，那分区数默认是1，直接return 0，不会启用自定义分区。

（97）自定义分区案例

首先抛出一个需求：将一堆手机号按照归属地的省份输出到不同的文件里。

已有一个phone_data.txt文件。

所以期望的输出数据是什么样子的呢？

手机号136/137/138/139开头的分别放进4个独立的文件里，然后其他的手机号放到一个文件里。最终形成5个文件。

显而易见，这个需求的核心在于自定义分区上。

所以我们需要写一个自定义分区类，假设它叫ProvincePartitioner，我们希望它能做到以下分配：

136 分区0
137 分区1
138 分区2
139 分区3
其他 分区4

等分区类建好后，别忘记在驱动里注册上这个类，并定义好ReduceTask数量。

job.setPartitionerClass(ProvincePartitioner.class);
job.setNumReduceTasks(5);

展示一下ProvincePartitioner类的代码：

package com.atguigu.mapreduce.partitioner;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;public class ProvincePartitioner extends Partitioner<Text, FlowBean> {@Overridepublic int getPartition(Text text, FlowBean flowBean, int numPartitions) {//获取手机号前三位prePhoneString phone = text.toString();String prePhone = phone.substring(0, 3);//定义一个分区号变量partition,根据prePhone设置分区号int partition;if("136".equals(prePhone)){partition = 0;}else if("137".equals(prePhone)){partition = 1;}else if("138".equals(prePhone)){partition = 2;}else if("139".equals(prePhone)){partition = 3;}else {partition = 4;}//最后返回分区号partitionreturn partition;}
}