九、Hadoop核心组件之MapReduce

左手的ㄟ右手 2021-09-01 06:14 366阅读 0赞

前面几篇文章我们呢着重介绍了HDFS，从这篇开始，我们开始介绍MapReduce，关注专栏[《破茧成蝶——大数据篇》][Link 1]查看相关系列的文章~

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

**目录**

一、MapReduce的定义

二、MapReduce的优缺点

2.1 优点

2.2 缺点

三、MapReduce的核心思想

四、MapReduce进程

五、编码实现MapReduce的WordCount

5.1 序列化类型

5.2 编程规范

5.3 实现WordCount

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

# 一、MapReduce的定义 #

MapReduce是一个分布式运算程序的框架，MapReduce的核心功能是将用户编写的业务逻辑代码和自带的默认组件整合成一个完整的分布式运算程序，并运行在Hadoop集群上。

# 二、MapReduce的优缺点 #

## 2.1 优点 ##

（1）MapReduce易于编程。它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的PC机器上运行。（2）良好的扩展性。（3）高容错性。（4）适合海量数据的离线处理。

## 2.2 缺点 ##

（1）不擅长实时计算。（2）不擅长流式计算。流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的，不能动态变化。（3）不擅长DAG（有向图）计算。多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写到磁盘，会造成大量的磁盘IO，导致性能非常的低下。

# 三、MapReduce的核心思想 #

（1）分布式的运算程序往往需要分成至少2个阶段。（2）第一个阶段的MapTask并发实例，完全并行运行，互不相干。（3）第二个阶段的ReduceTask并发实例互不相干，但是他们的数据依赖于上一个阶段的所有MapTask并发实例的输出。（4）MapReduce编程模型只能包含一个Map阶段和一个Reduce阶段，如果用户的业务逻辑非常复杂，那就只能多个MapReduce程序，串行运行。

# 四、MapReduce进程 #

一个完整的MapReduce程序在分布式运行时有三类实例进程：（1）MrAppMaster：负责整个程序的过程调度及状态协调。（2）MapTask：负责Map阶段的整个数据处理流程。（3）ReduceTask：负责Reduce阶段的整个数据处理流程。

简单来说，MapReduce由Map和Reduce两个阶段组成。Map阶段并行的处理输入的数据，Reduce阶段对Map的结果进行汇总。MapTask将结果写入磁盘，ReduceTask从每个MapTask上读取一份数据，Shuffle连接Map和Reduce两个阶段。

# 五、编码实现MapReduce的WordCount #

## 5.1 序列化类型 ##

通过分析源码发现，WordCount案例有Map类、Reduce类和驱动类。且数据的类型是Hadoop自身封装的序列化类型。常用的数据序列化类型如下所示：

![20200727134215285.png][]

## 5.2 编程规范 ##

编写的程序分为三部分：Mapper、Reducer和Driver。

**1、Mapper阶段**

（1）用户自定义的Mapper要继承自己的父类。（2）Mapper的输入数据是KV对的形式（KV类型可自定义）。（3）Mapper中的业务逻辑写在map()方法中。（4）Mapper的输出数据是KV对的形式（KV类型可以自己定义）。（5）map()方法（MapTask进程）对每个<K,V>调用一次。

** 2、Reducer阶段**

（1）用户自定义的Reducer要继承自己的父类。（2）Reducer的输入数据类型对应Mapper的输出数据类型，也是KV。（3）Reducer的业务逻辑写在reduce()方法中。（4）ReduceTask进程对每一组相同K的<K,V>组调用一次reduce()方法。

**3、Driver阶段**

相当于Yarn集群的客户端，用于提交我们整个程序到Yarn集群，提交的是封装了MapReduce程序相关运行参数的job对象。

## 5.3 实现WordCount ##

1、需求：给定一个文件，统计该文件中每个单词出现的次数。文件内容如下所示：

![20200727135647809.png][]

2、新建Maven工程，在pom.xml文件中添加如下依赖：

<dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-core</artifactId>
        <version>2.8.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>2.7.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>2.7.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>2.7.2</version>
    </dependency>

3、在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，在文件中填入如下内容：

log4j.rootLogger=INFO, stdout
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
    log4j.appender.logfile=org.apache.log4j.FileAppender
    log4j.appender.logfile.File=target/spring.log
    log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
    log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

4、编写Mapper类

package com.xzw.hadoop.mapreduce.wordcount;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    /**
     *                             _ooOoo_
     *                            o8888888o
     *                            88" . "88
     *                            (| -_- |)
     *                            O\  =  /O
     *                         ____/`---'\____
     *                       .'  \\|     |//  `.
     *                      /  \\|||  :  |||//  \
     *                     /  _||||| -:- |||||-  \
     *                     |   | \\\  -  /// |   |
     *                     | \_|  ''\---/''  |   |
     *                     \  .-\__  `-`  ___/-. /
     *                   ___`. .'  /--.--\  `. . __
     *                ."" '<  `.___\_<|>_/___.'  >'"".
     *               | | :  `- \`.;`\ _ /`;.`/ - ` : | |
     *               \  \ `-.   \_ __\ /__ _/   .-` /  /
     *          ======`-.____`-.___\_____/___.-`____.-'======
     *                             `=---='
     *          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
     *                     佛祖保佑        永无BUG
     * @Description
     * @Author xzw
     * @Date Created by 2020/5/19 10:16
     */
    public class WcMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        private Text word = new Text();
        private IntWritable one = new IntWritable(1);
    
        /**
         * MapTask进程
         * @param key
         * @param value
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //拿到这行数据
            String line = value.toString();
    
            //按照Tab进行切分
            String[] words = line.split("\t");
    
            //遍历数组，把单词变成(word,1)的形式
            for (String word: words) {
                this.word.set(word);
                context.write(this.word, this.one);
            }
        }
    }

5、编写Reducer类

package com.xzw.hadoop.mapreduce.wordcount;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /**
     * @Description
     * @Author xzw
     * @Date Created by 2020/5/19 10:27
     */
    public class WcReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable total = new IntWritable();
    
        /**
         * ReduceTask进程
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException,
                InterruptedException {
            //求和
            int sum = 0;
            for (IntWritable value: values) {
                sum += value.get();
            }
    
            //包装结果并输出
            total.set(sum);
            context.write(key, total);
    
        }
    }

6、编写Driver驱动类

package com.xzw.hadoop.mapreduce.wordcount;
    
    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 java.io.IOException;
    
    /**
     * @Description
     * @Author xzw
     * @Date Created by 2020/5/19 10:36
     */
    public class WcDriver {
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            //设置输入输出路径（便于测试）
            args = new String[]{"e:/input/xzw.txt", "e:/output"};
    
            //1、获取一个job实例
            Job job = Job.getInstance(new Configuration());
    
            //2、设置类路径
            job.setJarByClass(WcDriver.class);
    
            //3、设置Mapper和Reducer
            job.setMapperClass(WcMapper.class);
            job.setReducerClass(WcReducer.class);
    
            //4、设置Mapper和Reducer的输出类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            //5、设置输入输出数据（便于测试，直接调用自己定义的路径）
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
            //6、提交job
            boolean b = job.waitForCompletion(true);
            System.exit(b ? 0 : 1);
        }
    }

7、测试，得到测试结果

![20200727141335709.png][]

part-r-00000中的内容：

![20200727141403907.png][]

程序到这里就说明一个简单的WordCount就已经实现了。刚刚是在本地测试，也可以打包放到集群上运行。以下便为大家介绍如何打包提交到Yarn上运行（我明明记得在之前我整理过一篇打包的博客，里面写的很详细具体的打包过程，可惜实在是找不到了……抓狂中~~）。使用IDEA构建的Maven项目打包方式也是五花八门，有可以用Maven直接打包的，也有可以用Maven插件打包的，小编这里介绍一种原生的、非常朴素的却十分好用的打包方式，因为使用Maven打包可能会因为网络等问题，例如插件啥的没下载全导致打包失败，好了，废话不多说了，直接开始吧。

（1）File-->Project Structure-->Artifacts-->+-->JAR-->From modules with dependencies...

![20200727150356278.png][]

（2）选择主类

![20200727150459392.png][]

（3）点击Apply-->OK即可

![2020072715054523.png][]

（4）Build-->Build Artifacts

![2020072715061485.png][]

（5）在弹出的小框中选择Build

![20200727150654551.png][]

（6）将打好的包上传到服务器上，运行如下命令：

hadoop jar wordcount.jar com.xzw.hadoop.mapreduce.wordcount.WcDriver /xzw/xzw.txt /xzw/output

出现下图结果，说明运行成功：

![20200727150909235.png][]

（7）查看集群运行结果

![2020072715093778.png][]

至此，本文就讲解完了，你们在这个过程中遇到了什么问题，欢迎留言，让我看看你们遇到了什么问题~

[Link 1]: https://blog.csdn.net/gdkyxy2013/category_10006538.html
[20200727134215285.png]: /images/20210726/0b99013d25584138b04f7149773b90aa.png
[20200727135647809.png]: /images/20210726/222cba34260c43b9b43cfa30b56865eb.png
[20200727141335709.png]: /images/20210726/9cb0670b3ccf46babab85311af802e11.png
[20200727141403907.png]: /images/20210726/d5b9d581590e4ab39a8fea49d4f2b36b.png
[20200727150356278.png]: /images/20210726/426dcd22ac064e2e9a8fd3a021032acc.png
[20200727150459392.png]: /images/20210726/5bcbab5709a24345a75177beadbe1a02.png
[2020072715054523.png]: /images/20210726/92a3cb6d9648406c887ef5989ca47747.png
[2020072715061485.png]: /images/20210726/dd047d940e214287a417459c37058029.png
[20200727150654551.png]: /images/20210726/bc07c1096a7b494bac7ebddf854ad757.png
[20200727150909235.png]: /images/20210726/ce64c4f12d704fe39aaf3290dd694cac.png
[2020072715093778.png]: /images/20210726/56cf213c06f3477e906e1e88adc643f4.png