Hadoop-MapReduce-Shuffle工作机制，Partition分区源码解析-连载中

刺骨的言语ヽ痛彻心扉 2023-03-01 10:48 101阅读 0赞

### 文章目录 ###

*   *   *  Shuffle机制
         *  Partition分区源码解析
         *  Partition分区案例实操

### Shuffle机制 ###

Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]

### Partition分区源码解析 ###

分区: 
       1. 默认的分区器 HashPartitioner, 根据key的哈希值对reduceTask的个数 进行取余操作来计算分区.
        public int getPartition(K key, V value,
                              int numReduceTasks) {
        
        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
        }
    
      
       2. 如何获取分区器对象
         NewOutputCollector(org.apache.hadoop.mapreduce.JobContext jobContext,
                           JobConf job,
                           TaskUmbilicalProtocol umbilical,
                           TaskReporter reporter
                           ) throws IOException, ClassNotFoundException {
        
          collector = createSortingCollector(job, reporter);
          partitions = jobContext.getNumReduceTasks();  // 获取reduceTask的个数，在driver中设置的
          // 设置的NumReduceTasks个数大于1，才会走自定义分区，若不设置，默认只有一个分区
          if (partitions > 1) {
        
            //通过反射的方式创建分区器
            partitioner = (org.apache.hadoop.mapreduce.Partitioner<K,V>)
              ReflectionUtils.newInstance(jobContext.getPartitionerClass(), job);
          } else {
        
            partitioner = new org.apache.hadoop.mapreduce.Partitioner<K,V>() {
        
              @Override
              public int getPartition(K key, V value, int numPartitions) {
        
                return partitions - 1;   //返回固定的分区号 0
              }
            };
          }
        }
    	// 这个就是通过反射获取分区器对象的方法
    	// String PARTITIONER_CLASS_ATTR = "mapreduce.job.partitioner.class";
        public Class<? extends Partitioner<?,?>> getPartitionerClass() 
         throws ClassNotFoundException {
        
        return (Class<? extends Partitioner<?,?>>) 
          //获取 PARTITIONER_CLASS_ATTR 保存的分区器对象，若没有，则默认使用HashPartitioner
          conf.getClass(PARTITIONER_CLASS_ATTR, HashPartitioner.class);
         }
    
        3. 写出kv要计算kv对应的分区号
         @Override
        public void write(K key, V value) throws IOException, InterruptedException {
        
          // kv被收集到缓冲区时，要先把分区号计算出来。
          collector.collect(key, value,
                            partitioner.getPartition(key, value, partitions));
        }
    
        4. 自定义分区: 继承Partitioner类. 重写getPartition方法。
           a. 在Driver中设置使用的分区类， 并设置对应的ReduceTask个数.
              业务决定将来有多少个分区，但是分区必须要由ReduceTask来控制。
           b. 分区数要按照实际分区器中业务逻辑决定的分区数来来设置，
              若ReduceTask的个数设置为1 ,则自定义的分区器无法使用。 
    	  	  若ReduceTask的个数设置的比实际的分区数少，报错。
    	      若ReduceTask的个数设置的比实际的分区数多，不报错，但会有reduceTask没有数据可处理。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 3]

### Partition分区案例实操 ###

**1）需求**

将统计结果按照手机归属地不同省份输出到不同文件中（分区）

（1）输入数据（phone\_data.txt）

（2）期望输出数据

手机号**136、137、138、139开头**都分别放到一个**独立的4个文件中**，其他开头的放到一个文件中。

**2）需求分析**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 4]  
**3）在案例2.4的基础上，增加一个分区类**

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {
        
    	@Override
    	public int getPartition(Text key, FlowBean value, int numPartitions) {
        
    		// 1 获取电话号码的前三位
    		String preNum = key.toString().substring(0, 3);		
    		// 2 判断是哪个省
    		int partition;
            switch (prePhone) {
        
                case "136":
                    partition = 0;
                    break;
                case "137":
                    partition = 1;
                    break;
                case "138":
                    partition = 2;
                    break;
                case "139":
                    partition = 3;
                    break;
                default:
                    partition = 4;
            }
            return partition;
    	}
    }

**4）在驱动函数中增加自定义数据分区设置和ReduceTask设置**

public class FlowsumDriver {
        
    
    	public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
        
    
    		// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
    		args = new String[]{
        "e:/iutput","e:/output"};
    
    		// 1 获取配置信息，或者job对象实例
    		Configuration configuration = new Configuration();
    		Job job = Job.getInstance(configuration);
    
    		// 2 指定本程序的jar包所在的本地路径
    		job.setJarByClass(FlowsumDriver.class);
    
    		// 3 指定本业务job要使用的mapper/Reducer业务类
    		job.setMapperClass(FlowCountMapper.class);
    		job.setReducerClass(FlowCountReducer.class);
    
    		// 4 指定mapper输出数据的kv类型
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(FlowBean.class);
    
    		// 5 指定最终输出的数据的kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(FlowBean.class);
    
    		// --------8 指定自定义数据分区---------------
    		job.setPartitionerClass(ProvincePartitioner.class);
    
    		// --------9 同时指定相应数量的reduce task----------------
    		job.setNumReduceTasks(5);
    		
    		// 6 指定job的输入原始文件所在目录
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    		// 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
    		boolean result = job.waitForCompletion(true);
    		System.exit(result ? 0 : 1);
    	}
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]: /images/20230208/5dee11e8d4124e59b293a54295c04e26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]: /images/20230208/263d0eb52f5b49a3bec850313ceab227.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200724124322234.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200724124336133.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200724124417587.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70