Hadoop学习——MapReduce的组件及简单API（二）

ゞ浴缸里的玫瑰 2023-10-18 13:08 85阅读 0赞

接上一篇，[Hadoop学习——MapReduce的组件及简单API（一）][Hadoop_MapReduce_API]，上一篇写了`mapreduce`的四个组件中的两个组件`Mapper`、`Reducer`。剩下的还有两个`Partitioner`组件和`Combiner`组件。

##### 一、分区组件 #####

###### 1、分区的介绍 ######

`Partitioner`组件即分区组件。  
  `MapReduce`默认分区是采用的`HashPartitioner`类中的如下代码来进行分区的：

public int getPartition(K2 key, V2 value, int numReduceTasks) {
    	return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }

通过以上代码可以看到，当任意数首先与`int`类型的最大值进行`与运算`的时候，会得到它本身，即还是`key.hashCode()`，当该值再与`numReduceTasks`取余，因为默认`reduceTask`数量为1，因此默认情况下，这里永远等于0。  
  通过上边，如果我们要设置分区，**那只要将`reduceTask`数量设置成大于1，那分区数量自然而然等于`reduceTask`数量。**  
  上一篇里也写过设置分区，只需要在`Driver`里，加一行配置`job.setNumReduceTasks(int tasks);` 即可。

当设置了分区的时候，流程就会如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvaGFxaXlp_size_16_color_FFFFFF_t_70]

上图的大致过程如下解释：

首先，`MapTask`和`ReduceTask`即我们继承的`Mapper`和`Reducer`类，`MapTask`的数量与最终的分块个数是一致的，但`ReduceTask`默认为`1`个，当在`Driver`类里，设置为`2`个`ReduceTask`时，再对`/word`下的文件进行计算时，`ReduceTask`即会有两个，并且以上边的`HashPartitioner`代码片段），将计算结果也分为两个`part`文件，**分布到这两个文件**里。

###### 2、分区文件的合并 ######

如果要让这两个文件聚合，可以执行类似下边的命令，这样就会将`hdfs`里的`、word/result`目录下的`part`开头的文件合并为`result.txt`：

hadoop fs -getmerge /word/result ./result.txt

###### 3、自定义分区 ######

分区操作是`shuffle`操作中的一个重要过程，作用就是将`map`的结果按照规则分发到不同`reduce`中进行处理，从而按照分区得到多个输出结果。  
  `Partitioner`是分区的基类，如果需要定制`partitioner`也需要继承该类,写分区规则，并且需要在`Driver`里设置分区类。

**具体如下：**

package mrDay1.mapreduce.flow;
    
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    /**
     * 开发一个自定义的分区器
     * 实现的父类里面的两个泛型类型，第一个对应mapperTask的输出key类型，
     * 第二个对应mapperTask的输出value类型
     * 分区默认是有分区编号的，如果有三个分区，即0、1、2,
     *分区编号是从0开始的。
     */
    public class FlowPartitioner extends Partitioner<Text , Flow>{
        
    
    	@Override
    	public int getPartition(Text key, Flow value, int pNum) {
        
    
    		if("hello".equals(value.getName())) {
        
    			return 0;
    		} else if ("world".equals(value.getName())) {
        
    			return 1;
    		}else {
        
    			return 2;
    		}
    	}
    
    }

以上，即根据数据行是`hello` 还是`world`等，将结果分为3个区。最后，还需要在`driver`中设定分区，加入如下代码：

//设置分区个数为3个
    job.setNumReduceTasks(3);
    //设置自定义分区组件
    //如果不设定，默认用的是HashPartitioner，该组件会按照Mapper输出key的
    //HashCode值做分区，确保相同的key落在同一个分区
    job.setPartitionerClass(FlowPartitioner.class);

**注意：**  
  如果上边只给出`1个分区`，默认会将所有结果，写入到一个文件中去，不会报错。  
  但是，如果不是`1个分区`，写成`2个分区`，会报错，大概是找不到第`3个分区`的错误。  
  即，**`分区个数`必须大于等于自定义分区类中定义的个数，或者可以设置为1**。其他情况都会报错。如果大于的时候，则多出来的文件会为空。

重新执行`driver`之后，会看到三个结果文件，分别按照刚才的分区设置分布的结果。

##### 二、合并组件 #####

###### 1、合并的介绍 ######

`Combiner`组件即是合并组件。  
  默认是没有`combine`过程的,combine的作用是让合并工作在`MapTask`提前发生,可以减少`reduceTask`的合并负载,**使用combine机制，不能改变最后的结果,即写法跟后边的reducer内容是一样的。**  
  如下图所示，对于以下3个`maptask`分别计算自己的切块数据，假设得到相同的结果，如果没有合并的过程，那`reducetask`抓取数据后,根据相同key做聚合的时候，处理的数据就会更多，占用的内存就会也多。  
  但如果在每个`maptask`中，加上了下图的`Combine`,提前做一次合并，那就会减少了`reduceTask`的压力,也降低了内存消耗，同时减少之间网络宽带的传输。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvaGFxaXlp_size_16_color_FFFFFF_t_70 1]

###### 2、合并组件的写法 ######

首先`mapper`和`reducer`的写法无变化，再创建一个`Combiner`类。因为合并组件的写法必须要与`reducer`一致，这里也是要**继承`Reducer`类**即可。

package mrDay1.mapreduce.combine;
    
    import java.io.IOException;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    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 value : values) {
        
    		count = count + value.get();
    		}
    		context.write(key, new IntWritable(count));
    	}
    }

然后，在driver中引入`combine`组件：

package mrDay1.mapreduce.combine;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import mrDay1.mapreduce.combine.WordCountDriver;
    import mrDay1.mapreduce.combine.WordCountMapper;
    import mrDay1.mapreduce.combine.WordCountReducer;
    
    public class WordCountDriver {
        
    
    	public static void main(String[] args) throws Exception {
        
    		Configuration conf = new Configuration();
    		//获取job对象
    		Job job = Job.getInstance(conf);
    		//设置job方法入口的驱动类
    		job.setJarByClass(WordCountDriver.class);
    		//设置Mapper组件类
    		job.setMapperClass(WordCountMapper.class);
    		//设置mapper的输出key类型
    		job.setMapOutputKeyClass(Text.class);
    		//设置Mappper的输出value类型，注意Text的导包问题
    		job.setMapOutputValueClass(IntWritable.class);
    		//设置reduce组件类
    		job.setReducerClass(WordCountReducer.class);
    		//设置reduce输出的key和value类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    		
    		//设置combine组件类，如果不设定，默认是没有combine过程的
    		//combine的作用是让合并工作在MapTask提前发生
    		//可以减少reduceTask的合并负载
    		//使用combine机制，不能改变最后的结果
    		job.setCombinerClass(WordCountCombiner.class);
    		
    		//设置输入路径
    		FileInputFormat.setInputPaths(job, new Path("hdfs://hadoop01:9000/word"));
    		//设置输出结果路径，要求结果路径事先不能存在
    		FileOutputFormat.setOutputPath(job, new Path("hdfs://hadoop01:9000/word/result"));
    		//提交job
    		job.waitForCompletion(true);
    	}
    }

以上即是`MapReduce`的最后两个组件介绍。

[Hadoop_MapReduce_API]: https://blog.csdn.net/wohaqiyi/article/details/85277345
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvaGFxaXlp_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20181227160334441.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvaGFxaXlp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvaGFxaXlp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20181227184702493.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvaGFxaXlp,size_16,color_FFFFFF,t_70