Hadoop高手之路5-MapReduce分布式计算框架

MapReduce是Hadoop系统核心组件之一，它是一种可用于大数据并行处理的计算模型、框架和平台，主要解决海量数据的计算，是目前分布式计算模型中应用较为广泛的一种。

一、MapReduce概述

1.MapReduce核心思想

MapReduce的核心思想是“分而治之”。所谓“分而治之”就是把一个复杂的问题，按照一定的“分解”方法分为等价的规模较小的若干部分，然后逐个解决，分别找出各部分的结果，把各部分的结果组成整个问题的结果，这种思想来源于日常生活与工作时的经验，同样也完全适合技术领域。

MapReduce作为一种分布式计算模型，它主要用于解决海量数据的计算问题。使用MapReduce操作海量数据时，每个MapReduce程序被初始化为一个工作任务，每个工作任务可以分为Map和Reduce两个阶段。

• Map阶段：负责将任务分解，即把复杂的任务分解成若干个“简单的任务”来并行处理，但前提是这些任务没有必然的依赖关系，可以单独执行任务。
• Reduce阶段：负责将任务合并，即把Map阶段的结果进行全局汇总。

MapReduce就是“任务的分解与结果的汇总”。即使用户不懂分布式计算框架的内部运行机制，但是只要能用Map和Reduce思想描述清楚要处理的问题，就能轻松地在Hadoop集群上实现分布式计算功能。

2.MapReduce编程模型

MapReduce是一种编程模型，用于处理大规模数据集的并行运算。使用MapReduce执行计算任务的时候，每个任务的执行过程都会被分为两个阶段，分别是Map和Reduce，其中Map阶段用于对原始数据进行处理，Reduce阶段用于对Map阶段的结果进行汇总，得到最终结果。

MapReduce编程模型借鉴了函数式程序设计语言的设计思想，其程序实现过程是通过map（）函数和reduce（）函数来完成。从数据格式上来看，map函数接受的数据格式是键值对，产生的输出结果也是键值对形式，reduce会将map函数的输出的键值对作为输入，把相同的key值的value进行汇总，输出新的键值对。

说明：对于某些任务来说，可能不一定需要Reduce，MapReduce的数据流模型坑内只有Map过程，由Map产生的数据直接写入HDFS。但是对于大多数的任务来说，都是需要Reduce的，而且可能需要多个reduce。

3.MapReduce编程实例——词频统计

首先，MapReduce通过默认组件TextInputFormat将待处理的数据文件（如text1.txt和text2.txt），把每一行的数据都转变为<key，value>键值对。

其次，调用Map()方法，将单词进行切割并进行计数，输出键值对作为Reduce阶段的输入键值对。

最后，调用Reduce()方法将单词汇总、排序后，通过TextOutputFormat组件输出到结果文件中。

二、MapReduce的工作原理

MapReduce框架要是由Map和Reduce阶段来实现计算的，那么这两个阶段内部是如何协同工作的呢？

1. MapReduc的工作过程

分为五个步骤

• 分片、格式化数据源
• 执行MapTask
• 执行Shuffle过程
• 执行ReduceTask
• 写入文件

1) 分片、格式化数据源

• 分片操作：分片将源文件划分为大小相等的小数据块（默认是128MB），Hadoop为每一个分片建立一个Map任务，并由该任务运行自定义的map函数，从而处理分片里的每一个记录。
• 格式化操作：将划分好的分片（split）格式化为键值对<Key,Value>形式的数据，key代表偏移量，value代表每一行内容

2) 执行MapTask

每个map任务都有一个内存缓冲区（缓冲区的大小为100MB），输入的分片数据经过Map任务处理后的中间结果会写入内存缓冲区。

3) 执行Shuffle

Shuffle会将MapTask输出的处理结果分发给ReduceTask，并在分发的过程中，对数据按key进行分区和排序。

4) 执行ReduceTask

输入ReduceTask的数据流是<key,{value list}>形式，用户可以自定义reduce方法进行逻辑处理，最终以<key,value>的形式输出。

5) 写入文件

MapReduce框架会自动把ReduceTask生成的<key,value>传入OutputFormat的write方法，实现文件的写入操作。

2. MapTask工作原理

MapTask作为MapReduce工作流程前半部分，它主要经历5个阶段，分别是Read阶段、Map阶段、Collect阶段、Spill阶段和Combiner阶段。

1) Read 阶段

通过用户编写的RecordReader，从输入的InputSplit中解析出一个个的key/value

2) Map阶段

将解析出的key/value交给用户编写的map函数处理，并产生一系列的新的key/value

3) Collect阶段

将会生成的key/value分片，写入一个环形的缓冲区中

4) Spill阶段

即“溢写”，当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。

5) Combine阶段

当所有数据处理完成后，MapTask会对所有的临时文件进行一次合并，以确保最终只会生成一个数据文件。

3. ReduceTask工作原理

ReduceTask的工作过程主要经历了5个阶段，分别是Copy阶段、Merge阶段、Sort阶段、Reduce阶段和Write阶段。

1) Copy阶段

Reduce会从各个MapTask上远程复制一片数据，并针对某一片数据，如果其大小超过一定的阈值，则写到磁盘上，否则直接放到内存中。

2) Merge阶段

在远程复制的同时，ReduceTask会启动两个后台进程，分别对内存和磁盘上的文件进行合并，以防止内存或者磁盘文件使用过多。

3) Sort阶段

为了将key相 同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己处理结果进行了局部排序，因此ReduceTask只需对所有数据进行一个归并排序。

4) Reduce阶段

对排序后的键值对调用reduce方法，最后把这些输出的键值对写入到HDFS中

5) Write阶段

reduce将计算结果写的HDFS上

4. Shuffle工作原理

Shuffle是MapReduce的核心，它用来确保每个reducer的输入都是按键排序的。它的性能高低直接决定了整个MapReduce程序的性能高低，map和reduce阶段都涉及到了shuffle机制。

三、MapReduce编程组件

1.InputFormat组件

主要用于描述输入数据的格式，它提供两个功能，分别是数据切分和为Mapper提供输入数据。

2.Mapper组件

Hadoop提供的Mapper类是实现Map任务的一个抽象基类，该基类提供了一个map()方法。

3.Reducer组件

Map过程输出的键值对，将由Reducer组件进行合并处理，最终的某种形式的结果输出。

4.Partitioner组件

Partitioner组件可以让Map对Key进行分区，从而可以根据不同的key分发到不同的Reduce中去处理，其目的就是将key均匀分布在ReduceTask上

5.Combiner组件

Combiner组件的作用就是对Map阶段的输出的重复数据先做一次合并计算，然后把新的（key，value）作为Reduce阶段的输入。

6.OutputFormat组件

OutputFormat是一个用于描述MapReduce程序输出格式和规范的抽象类。

四、MapReduce运行模式与性能优化策略

1.运行模式

1) 本地运行模式

在当前的开发环境模拟MapReduce执行环境，处理的数据及输出结果在本地操作系统。

2) 集群运行模式

把MapReduce程序打成一个Jar包，提交至Yarn集群上去运行任务。由于Yarn集群负责资源管理和任务调度，程序会被框架分发到集群中的节点上并发的执行，因此处理的数据和输出结果都在HDFS文件系统中。

2.性能优化策略

使用Hadoop进行大数据运算，当数据量极其大时，那么对MapReduce性能的调优重要性不言而喻，尤其是Shuffle过程中的参数配置对作业的总执行时间影响特别大，我们可以从五个方面对MapReduce程序进行性能调优，分别是数据输入、Map阶段、Reduce阶段、Shuffle阶段和其他调优属性方面。

1) 数据输入

在执行MapReduce任务前，将小文件进行合并，大量小文件会产生大量的map任务，增大map任务装载次数，而任务装载较耗时，从而导致MapReduce运行速度较慢。因此采用CombineTextInputFormat来作为输入，解决输入端大量的小文件场景。

2) Map阶段

• 减少溢写（spill）次数
• 减少合并（merge）次数
• 在map之后，不影响业务逻辑前提下，先进行combine处理，减少 I/O

3) Reduce阶段

• 合理设置map和reduce数
• 设置map、reduce共存
• 规避使用reduce
• 合理设置reduce端的buffer

4) Shuffle阶段

Shuffle阶段的调优就是给Shuffle过程尽量多地提供内存空间，以防止出现内存溢出现象，可以由参数mapred.child.java.opts来设置，任务节点上的内存大小应尽量大。

5) 其他调优属性

MapReduce还有一些基本的资源属性的配置，这些配置的相关参数都位于mapred-default.xml文件中，我们可以合理配置这些属性提高MapReduce性能，例如合理设置MapTask、ReduceTask等参数。

五、本地模式执行MapReduce

单词计数为案例

1. 启动idea，新建一个maven项目

2. 添加pom依赖

3. 新建包

4. 新建mapper类

输入代码

5. 新建reducer类，输入代码

6. 新建一个主类

7.运行，查看结果

查看结果文件夹

8. 添加一个日志输出文件

六、集群模式执行MapReduce

1. 启动hadoop集群

2. 项目通过本地模式进行调试，确认运行正确

3. 保持mappper和reducer类不变，跟本地模式一致

4. 修改主类

该行注释

去掉盘符

5. 准备好用于单词计数的文本文件并上传hadoop集群

在根目录下创建data文件夹。

将本地文件上传到虚拟机上。

将虚拟机上的文件上传到集群上。

在集群上出现了文件。

6. 项目打成jar包

7. 把jar包上传的linux虚拟机hadoop001下

8. 通过集群模式运行jar包

获取主类的全合格类名

9. 查看结果

七、MapReduce的经典案例一-数据去重

1. 案例分析

数据清洗一个重要的操作就是数据去重，也就是去除重复的数据。

2. 准备好数据文件

3. 需求

对这两个文件的额内容进行去重操作，并将最终的结果合并为一个文件。

• map阶段：将每一行作为key，value都可以设置为null
• reduce阶段：不需要考虑每一个key有多少个value，把value设置为null就可以自动去重

4. 编写mapper类

5. 编写reducer类

6. 编写主类

7. 本地模式执行查看结果

八、MapReduce的经典案例之二-TopN

1. 案例分析

TopN分析法是指从研究对象中按照某一个指标进行倒序或正序排列，取出排名靠前的N个数据，并对这N个数据进行重点分析，这种方法在大数据分析中经常会用到。

2. 准备好数据文件

3. 需求分析

提取上述文本文件中的最大的5个数，并将最终结果汇总到一个文件中

TreeMap

4.新建一个maven项目

添加依赖

新建包

5. 编写Mapper类

6. 编写reduce类

7. 编写主类

8. 运行，查看结果