MapReduce中的Combiner

短命女 2022-12-05 05:13 219阅读 0赞

## 前言 ##

MapReduce框架使用Mapper将数据处理成一个个的<key, value>键值对，然后在网络节点间对其进行整理（shuffle），最后使用Reducer处理数据并输出。

## 分析 ##

从上面的过程中，我们可以分析出两个性能瓶颈：

*  网络传输压力：比如我们有100亿条数据，Mapper就会生成100亿条键值对在网络间传输。如果我们只是对数据求最大值，那么很明显每一个Mapper只需要输出它所处理的数据中的最大值即可。这样不仅能减少网络传输压力，同时也能提升程序处理效率。
 *  数据倾斜：如果Mapper处理的数据有着严重的数据倾斜问题，那么会导致大部分数据汇聚到一个reduce中处理，导致reduce处理效率低甚至崩溃等情况。

## 目的 ##

MapReduce中的Combiner就是为了缓解Map任务和Reduce任务之间的数据传输问题的。MapReduce运行用户针对Map Task的输出设置一个合并函数来减少传输到reduce的数据，可以理解为map任务的本地reduce操作。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70][]

## 数据格式转换 ##

*  Mapper：（K1, V1）—> list(K2, V2)
 *  Combiner：（K2, list(V2)) —> list(K3, V3)
 *  Reducer：（K3, list(V3)）—> list(K4, V4)

**注意：**Combiner的输入与Reducer的输入一致，输出与Mapper的输出一致

## 使用注意 ##

*  Combiner与Mapper的数据合并不一样，Mapper的数据合并发生在数据spill的时候，即进行数据merge操作（具体机制可参考[《Hadoop中的MapTask工作机制》][Hadoop_MapTask]）
 *  Combiner没有默认的实现，需要显式调用才行
 *  并不是所以的Job都适合使用Combiner，只有数据满处满足结合律才能使用。比如数据求和、求最大值等，但求中值就不适用
 *  一般来说，Combiner与Reducer进行相同的操作，可以看做Combiner是本地reduce（可直接复用reduce处理逻辑或自定义实现）
 *  Combiner是针对每一个Map Task的，所以它的输入也只能是单个Map Task的输出

## 举例 ##

说了这么多，直接上示例，以单词统计为例

数据文件有3个

Hello World
    Hello Tom
    Hello Jerry
    Hello Jack

Hello LiLei
    Hello HanMeimei
    Tom and Jerry
    Welcome To China

Hello Jack
    China

程序代码如下

import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    import java.util.StringTokenizer;
    
    public class WordCountMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    
        private Text outputKey;
        private LongWritable outputVal;
    
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            outputKey = new Text();
            outputVal = new LongWritable(1);
        }
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer st = new StringTokenizer(value.toString());
            while (st.hasMoreTokens()) {
                outputKey.set(st.nextToken());
                context.write(outputKey, outputVal);
            }
        }
    }

import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    public class WordCounntReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    
        private Text outputKey;
        private LongWritable outputVal;
    
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            outputKey = new Text();
            outputVal = new LongWritable();
        }
    
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long sum = 0;
            for (LongWritable val: values) {
                sum += val.get();
            }
    
            outputKey = key;
            outputVal.set(sum);
            context.write(outputKey, outputVal);
        }
    }

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    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 org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    public class WordCountJob extends Configured implements Tool {
    
        @Override
        public int run(String[] strings) throws Exception {
    
            Configuration conf = getConf();
    
            String input = conf.get("input");
            String output = conf.get("output");
    
            Job job = Job.getInstance(conf);
            job.setJarByClass(WordCountJob.class);
            job.setMapperClass(WordCountMapper.class);
            job.setReducerClass(WordCounntReducer.class);
    
            // 显式设置Combiner
            job.setCombinerClass(WordCounntReducer.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);
    
            FileInputFormat.addInputPath(job, new Path(input));
            FileOutputFormat.setOutputPath(job, new Path(output));
    
            boolean rt = job.waitForCompletion(true);
            return rt? 0: 1;
        }
    
        public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            System.exit(ToolRunner.run(conf, new WordCountJob(), args));
        }
    }

在未设置Combiner时，结果如下:

*  Map输入条数为10条，输出条数为21条，每个单词生成一个<key, value>键值对；
 *  Combiner没有显式设置，故输入输出量为0
 *  Reduce输入条数为21条，输出条数为11条

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70 1][]

设置Combiner时，结果如下：

*  Map输入条数为10条，输出条数为21条，每个单词生成一个<key, value>键值对；
 *  Combiner显式设置，输入条数为21条，输出条数为17条，同一个map中相同的key被合并为一个key了；
 *  Reduce输入条数为17条，输出条数为11条

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70 2][]

参考：[https://blog.csdn.net/zpf336/article/details/82181304][https_blog.csdn.net_zpf336_article_details_82181304]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70]: /images/20221123/87e95ee3e1134a7d87ce037e8f43c611.png
[Hadoop_MapTask]: https://blog.csdn.net/YF_Li123/article/details/108486272
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70 1]: /images/20221123/3d1878b05d354e619c22e4c0ba4ba562.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1lGX0xpMTIz_size_16_color_FFFFFF_t_70 2]: /images/20221123/7249e3e3f5374effafee3ae9288a8ddc.png
[https_blog.csdn.net_zpf336_article_details_82181304]: https://blog.csdn.net/zpf336/article/details/82181304