Shuffle阶段：partition分区以及自定义使用注意事项

野性酷女 2022-06-04 10:18 227阅读 0赞

** 一：partition分区(shuffle阶段)**

Mapreduce中会将map输出的kv对，按 **照相同key分组** ， 然后分发给不同的reducetask（所以这也决定了为什么最终的文件个数，即分区个数跟reducetask数量一样了。） 。默认分区是根据key的hashCode对reduceTasks个数取模得到的。用户没法控制哪个key存储到哪个分区。默认系统的patitioner类，实现类时hashpatitioner.

**默认的分发规则为：根据key的hashcode%reducetask数来分发**

所以：如果要按照我们自己的需求进行分组，则需要改写数据分发（分组）组件Partitioner

自定义一个CustomPartitioner继承抽象类：Partitioner

然后在job对象中，设置自定义partitioner： job.setPartitionerClass(CustomPartitioner.class)

1）默认partition分区：要从0开始，否则报错

public class HashPartitioner<K, V> extends Partitioner<K, V> \{

/\*\* Use \{@link Object\#hashCode()\} to partition. \*/

public int getPartition(K key, V value, int numReduceTasks) \{

**  return (key.hashCode() & Integer.MAX\_VALUE) % numReduceTasks;**

//一般都是分区个数自动设置，通过上面的计算结果

注意： **重写的mypatition方法继承了抽象类patitioner。然后在getPartition方法中重新定义分区的条件，用什么进行分区。默认hashpatitioner传入的分区个数是等于reducetask的个数的，所以传入的分区的个数等于reducetask个数。  **

**      自定义的分区则是我们想怎么定义怎么定义，然后再根据我们的分区的条件产生的分区个数进行设置reducetask个数。而且要求设置reducetask最好要等于分区个数** 。

** 所以如果我们无论怎么定义分区，而reducetASK的任务始终设置成为1个，则没有任何意义，因为最终的分区文件还是由reducetask的个数决定的。**

2.自定义分区

**1.自定义类继承Partitioner，泛型声明对谁排序。重写getPartition()方法**

** ** **public**   **class**  ProvincePartitioner  **extends**   Partitioner <Text, FlowBean> \{

// **key是要分区的字段，value分区要键入的值，KV也是MAP输出的key,value.**

**     //return 返回的结果值就是分区号**

@Override

**public**   **int**  getPartition(Text key, FlowBean value,  **int**  numPartitions) \{

// 这里的 numpationns 就是分区的个数，也就是对应的 reduce 要开启的

Reducetask 的个数。

// 1  获取电话号码的前三位

String preNum = key.toString().substring(0, 3);

// 手动设置分区个数

**int**  partition = 4;

// 2  判断是哪个省

**if**  ("136".equals(preNum)) \{

partition = 0;

\} **else**   **if**  ("137".equals(preNum)) \{

partition = 1;

\} **else**   **if**  ("138".equals(preNum)) \{

partition = 2;

\} **else**   **if**  ("139".equals(preNum)) \{

partition = 3;

**return**  partition;

--------------------

2 ） 在 job 驱动中，设置自定义 partitioner ：

job.setPartitionerClass(CustomPartitioner.class)

3） 自定义 partition 后，要根据自定义 partitioner 的逻辑设置相应数量的 reduce task

job.setNumReduceTasks(5);

注意：

如果 reduceTask 的数量 >= getPartition 的结果数，则会多产生几个空的输出文件 part-r-000xx ；

如果 1<reduceTask 的数量 <getPartition 的结果数，则有一部分分区数据无处安放，会 Exception ；

如果 reduceTask 的数量 =1 ，则不管 mapTask 端输出多少个分区文件，最终结果都交给这一个 reduceTask ，最终也就只会产生一个结果文件  part-r-00000 ；（默认也是hashpatitioner分区，只是最终分区到同一文件里了，看不出来）

注意： Patitionner 分区实际是通过设置 reducetask 的数量来产生文件的数量。而分区 Patitionner 是通过该方法的返回值确定的

例如：假设自定义分区数为 5 ，则

（ 1 ） job.setNumReduceTasks(1); 会正常运行，只不过会产生一个输出文件

（ 2 ） job.setNumReduceTasks(2); 会报错

（ 3 ） job.setNumReduceTasks(6); 大于 5 ，程序会正常运行，会产生空文件

） 定义的 patitions  必须从 0 开始，所以如果定义了 patition = 4. 则， NumReduceTaskSy=5

3.分区partition案例

把单词按照首字母ASCII码奇偶分区（Partitioner）

package com.robot .mapreduce.wordcount;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Partitioner;

public class WordCountPartitioner extends Partitioner<Text, IntWritable>\{

@Override

public int getPartition(Text key, IntWritable value, int numPartitions) \{

**//注意 ，如果在定义的类中使用HashPartition进行分区，要重写hashcode方法**

// 1 获取单词key

String firWord = key.toString().substring(0, 1);

//直接用下面这一句也可以，自动类型转换

//char  firWord = key.toString().charAt(0);

//转换成ASCII码

char\[\] charArray = firWord.toCharArray();

int result = charArray\[0\];

// 2 根据奇数偶数分区

if (result % 2 == 0) \{

return 0;

\}else \{

return 1;

2）在驱动中配置加载分区，设置reducetask个数

job.setPartitionerClass(WordCountPartitioner.class);

job.setNumReduceTasks(2);

Shuffle阶段：partition分区以及自定义使用注意事项

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 MapReduce自定义分区Partition

相关 [Protobuf]安装与使用以及注意事项

相关使用 nvue 的注意事项以及坑

相关 oracle安装教程以及使用注意事项

相关 java static关键字以及使用注意事项

相关自定义函数，函数的使用注意事项，自定义对象，对象直接量

相关 Shuffle阶段：partition分区以及自定义使用注意事项

相关 Hibernate的使用以及注意事项

相关 vuelidate框架自定义验证规则【注意事项】

相关 hive 分区表使用的一些注意事项