Hadoop MR 之(三) Combiner类 / Partitioner 类 / GroupingComparator 类

小灰灰 2022-02-23 03:22 228阅读 0赞

### 前言 ###

在前一章中, 我们介绍了如何使用`MR框架`完成自己的业务逻辑. 并且,实现了排序功能. 本章, 我们继续讲解Hadoop MR的自定义处理细节.

本文相关代码, 可在我的Github项目 [https://github.com/SeanYanxml/bigdata/][https_github.com_SeanYanxml_bigdata] 目录下可以找到. PS: (如果觉得项目不错, 可以给我一个Star.)

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

### 基础知识 ###

在讲解`Hadoop MR`的`Combainor`/`Partitioner`/GroupingComparator之前, 我们先梳理下`MR`的处理逻辑. 以便于我们后续的讲解工作.

MR的处理逻辑如下所示(本过程通常也被成为`Hadoop Shuffle`操作):  
图暂略(后补)  
处理的逻辑主要包括如下几个部分

##### Mapper处理逻辑 #####

*  通过InputFormat类读取数据(读取文件使用`FileInputFormat`)
 *  通过`Combiner`将Key值相同的数据项合并.(Ex `<hello,1><hello,1>`合并为<hello,2>)
 *  通过`Partitioner`指定所被分配的`Reduce Task`的ID.
 *  随后, 将转换好的数据放入`context`内部, 交与`Reduce Task`进行处理.

##### Reducer处理逻辑 #####

*  读取Reducer交付的数据.
 *  `setup()`/`map()`/`cleanup()` 运行各自的逻辑.
 *  通过OutputFormat进行输出.(默认为`FileOutputFormat`)

本章我们主要讲述自定义化`Combiner`/`Partitioner`/ `GroupingComparator`的部分.

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

### 正文 ###

##### 自定义Combiner类 #####

自定义`Combainor` 的举例中, 我们列举在计算WordCount时将`<hello,1><hello,1>`合并为`<hello,2>`的操作.

public class WordCountCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{
    
    	@Override
    	protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context)
    			throws IOException, InterruptedException {
    		int count = 0;
    		for(IntWritable v: values){
    			count += v.get();
    		}
    		context.write(key, new IntWritable(count));		
    	}
    }
    //  随后在 Job方法内进行指定方法即可完成.
    //

随后在 Job方法内进行指定.`(job.setCombinerClass(WordCountCombiner.class);)`.值得一提的是, 使用`Combiner`类型的方法的前提是不改变原有的数据逻辑. (比如: (2+1)/2 (2+1)/2 就不是适合合并. 合并后结果为(2+1+2+1)/2 结果为3, 原逻辑结果为2.)

##### 自定义Partitioner 类 #####

自定义`Partitioner`类中, 我们以Flow流量的自定义切分作为例子.(前面的例子, 详见[Hadoop MR 之(二) 传递自定义数据类型 & 排序][Hadoop MR _ _ _])

public class FlowBeanPartitioner extends Partitioner<FlowBean, NullWritable>{
    
    	// numPartition为reducetask的个数
    	@Override
    	public int getPartition(FlowBean bean, NullWritable value, int numPartition) {
    		// 相同id订单bean, 会发往相同的partition.
    		// 而且, 产生的分区数, 会根用户设置的reduce task保持一致.
    		return (bean.getPhoneNumber().hashCode() & Integer.MAX_VALUE) % numPartition;
    	}
    }

通过Job内进行指定, 即可完成. `job.setPartitionerClass(FlowBeanPartitioner.class);`.  
此处有一个小细节. `numPartition`为指定的`Reducer Task`的个数, 当个数超过输出值的时候, 会出现异常.  
(如: 返回值为`4`, 但是`Reducer Task`个数设置为2, 就会报异常. 如果返回值为`1-4`,`Reducer Task`个数为8, 那么就会出现某几个`Reducer`未分配到任务的情况. 也就是`Reducer`空启,未处理任何数据. )

##### 自定义GroupingComparator 类 #####

自定义`Partitioner`类中, 我们仍然Flow流量的自定义切分作为例子.(前面的例子, 详见[Hadoop MR 之(二) 传递自定义数据类型 & 排序][Hadoop MR _ _ _]).

在排序的时候我们传入的数据类型为`<bean,value>`类型, 为了使作为`<key>`位置的`FlowBean`进行合并. 但是在处理时候`<188,xx,xx>` `<188,xx,xx>`其实可以合并的,但是因为此时的`<key>`值为`Bean`的不同对象.所以系统默认认为两个不相同, 我们可以自定义比较函数, 用于`判别两个对象是否相同?`.

public class FlowBeanComparator extends WritableComparator{
    	// 传入作为key的class类型. 让框架做反射作为实例对象.
    	protected FlowBeanComparator(){
    		// 用于注册某个用于排序的bean.
    		super(FlowBean.class,true);
    	}
    
    	// 
    	@Override
    	public int compare(Object a, Object b) {
    		FlowBean aBean = (FlowBean) a;
    		FlowBean bBean = (FlowBean) b;
    		// 比较两个bean的 指比较bean内的orderid
    		return aBean.getPhoneNumber().compareTo(bBean.getPhoneNumber());
    	}
    }

随后可以通过设置 `// 设置相同对象的判断器 job.setGroupingComparatorClass(FlowBeanComparator.class);`让其生效.

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

[https_github.com_SeanYanxml_bigdata]: https://github.com/SeanYanxml/bigdata/
[Hadoop MR _ _ _]: https://blog.csdn.net/u010416101/article/details/89050417