系列文章目录

一、MapReduce概述
二、MapReduce工作机制
三、Map，Shuffle，reduce阶段详解
四、大案例解析

---

---

前言

MapReduce（简称MR）是Hadoop框架中的一个核心组件，它主要用于大规模数据集的并行处理。MapReduce的底层原理涉及数据的分片、Map阶段的处理、Shuffle过程以及Reduce阶段的处理等多个环节。本文通过案例解析方式对MR底层原理进行详细解析

---

一、MapReduce概述

MapReduce是一种编程模型和相关的实现，用于大规模数据集（大于1TB）的并行运算。它由Google在2004年提出，随后Apache Hadoop项目实现了MapReduce的开源版本。MapReduce将复杂的运行于大规模集群上的并行计算过程高度地抽象到了两个函数：Map和Reduce。Map函数处理输入数据并产生中间键值对，Reduce函数则对中间键值对进行合并以产生最终结果。

二、MapReduce工作机制

1. 角色与组件

MapReduce作业的执行涉及多个角色和组件，主要包括：

Client：提交MapReduce作业的客户端。
ResourceManager（RM）：负责集群资源的统一管理和调度。
NodeManager（NM）：负责节点上的资源管理和使用，并定时向RM汇报资源使用情况。
ApplicationMaster（AM）：每个作业都会有一个AM，负责应用程序的管理，包括资源的请求、任务的分配和监控等。
HDFS：Hadoop分布式文件系统，用于存储作业的输入数据和输出数据。

2. 作业提交与执行流程

1. 作业提交：

Client将作业提交给ResourceManager。
ResourceManager中的Scheduler为作业分配资源，并启动ApplicationMaster。
ApplicationMaster根据作业需求向ResourceManager申请资源，并启动MapTask和ReduceTask。

2. Map阶段：

MapTask从HDFS读取输入数据（以split为单位）。
每个MapTask处理一个split，将其中的数据转换为键值对（Key/Value）。
对每个键值对执行Map函数，输出中间结果（也是键值对形式）。
中间结果会暂时存储在本地磁盘的环形缓冲区中，当缓冲区满或Map任务完成时，会将数据写入HDFS中的临时目录。

3. Shuffle阶段：

Shuffle是MapReduce的核心过程之一，它负责将Map任务的输出作为Reduce任务的输入。
Shuffle过程包括Collect、Spill、Merge、Copy和Sort等阶段。
Collect阶段：MapTask将结果输出到环形缓冲区。
Spill阶段：当缓冲区满时，将数据写入本地磁盘，并进行排序和合并。

4. Merge阶段：将所有溢出的临时文件合并成一个文件。

Copy阶段：ReduceTask启动Fetcher线程从MapTask所在节点复制属于自己的数据。
Sort阶段：ReduceTask对复制过来的数据进行排序，确保每个key对应的value值都排在一起。

5. Reduce阶段：

ReduceTask从Shuffle阶段获取到排序后的数据。
对每个key及其对应的value列表执行Reduce函数，生成最终结果。
将最终结果写入HDFS。

三、Map阶段详解

1. 数据分片（Split）

输入源：MapReduce作业通常以HDFS上的文件作为输入源，但也可以有其他输入源。
文件分片：HDFS上的文件被逻辑上划分为多个分片（split），每个split包含一个或多个block（块），默认是一对一的关系。split不包含具体数据，只包含数据的位置信息。
分片大小：分片大小通常等于HDFS的block大小（默认128MB），但可以通过配置参数进行调整。

2. Map函数执行

输入：Map函数以split中的数据作为输入，数据被转换为键值对形式。对于HDFS文件，键通常是数据的偏移量，值是数据本身。
处理：用户自定义的Map函数对每个键值对进行处理，并输出中间结果（也是键值对形式）。
输出：中间结果被写入到本地磁盘的环形缓冲区中，当缓冲区满或Map任务完成时，会被写入HDFS中的临时目录。

四、Shuffle阶段详解

Shuffle阶段是MapReduce作业中最复杂也是最关键的过程之一，它负责将Map任务的输出数据整理并传递给Reduce任务。

1. Collect阶段

MapTask将处理后的数据写入到环形缓冲区中，同时保存分区信息和序列化后的键值对数据。

2. Spill阶段

当环形缓冲区中的数据量达到一定的阈值（如80%满）时，会触发Spill操作。
Spill操作将缓冲区中的数据排序并写入到本地磁盘的临时文件中。如果配置了Combiner，还会对相同分区和key的数据进行合并。

3. Merge阶段

当MapTask完成后，会将所有溢出的临时文件合并成一个大的文件。
合并

五、大案例解析

1. 单词统计流程

已知文件内容:

    hadoop hive hadoop spark hive flink hive linux hive mysqlinput结果: k1(行偏移量)   v1(每行文本内容)0   		 hadoop hive hadoop spark hive 30   		 flink hive linux hive mysql
map结果:k2(split切割后的单词)  v2(拼接1)     hadoop  			 1hive   				 1hadoop  			 1spark  				 1hive   				 1flink  				 1hive   				 1linu  				 1hive   				 1mysql  				 1
分区/排序/规约/分组结果:k2(排序分组后的单词)   v2(每个单词数量的集合)flink   		    [1]hadoop  			[1,1]hive    			[1,1,1,1]linux   			[1] mysql   			[1]spark   			[1]
reduce结果:k3(排序分组后的单词)   v3(聚合后的单词数量)flink 				 1hadoop 				 2hive   				 4linux  				 1mysql  				 1spark  				 1
output结果:   注意: 输出目录一定不要存在,否则报错flink   1hadoop  2hive    4linux   1mysql   1spark   1

2. MR底层原理图解析

2.1 map阶段流程

第一阶段是把输入目录下文件按照一定的标准逐个进行逻辑切片，形成切片规划。默认情况下Split size 等于 Block size。每一个切片由一个MapTask处理（当然也可以通过参数单独修改split大小）
第二阶段是对切片中的数据按照一定的规则解析成对。默认规则是把每一行文本内容解析成键值对。key是每一行的起始位置(单位是字节)，value是本行的文本内容。（TextInputFormat）
第三阶段是调用Mapper类中的map方法。上阶段中每解析出来的一个，调用一次map方法。每次调用map方法会输出零个或多个键值对
第四阶段是按照一定的规则对第三阶段输出的键值对进行分区。默认是只有一个区。分区的数量就是Reducer任务运行的数量。默认只有一个Reducer任务
第五阶段是对每个分区中的键值对进行排序。首先，按照键进行排序，对于键相同的键值对，按照值进行排序。比如三个键值对<2,2>、<1,3>、<2,1>，键和值分别是整数。那么排序后的结果是<1,3>、<2,1>、<2,2>。
如果有第六阶段，那么进入第六阶段；如果没有，直接输出到文件中
第六阶段是对数据进行局部聚合处理，也就是combiner处理。键相等的键值对会调用一次reduce方法。经过这一阶段，数据量会减少。本阶段默认是没有的。

2.2 shuffle阶段流程

shuffle是Mapreduce的核心，它分布在Mapreduce的map阶段和reduce阶段。一般把从Map产生输出开始到Reduce取得数据作为输入之前的过程称作shuffle。
Collect阶段：将MapTask的结果输出到默认大小为100M的环形缓冲区，保存的是key/value，Partition分区信息等
Spill阶段：当内存中的数据量达到一定的阀值(80%)的时候，就会将数据写入本地磁盘，在将数据写入磁盘之前需要对数据进行一次排序的操作，如果配置了combiner，还会将有相同分区号和key的数据进行排序
Merge阶段：把所有溢出的临时文件进行一次合并操作，以确保一个MapTask最终只产生一个中间数据文件
Copy阶段： ReduceTask启动Fetcher线程到已经完成MapTask的节点上复制一份属于自己的数据，这些数据默认会保存在内存的缓冲区中，当内存的缓冲区达到一定的阀值的时候，就会将数据写到磁盘之上
Merge阶段：在ReduceTask远程复制数据的同时，会在后台开启两个线程对内存到本地的数据文件进行合并操作。
Sort阶段：在对数据进行合并的同时，会进行排序操作，由于MapTask阶段已经对数据进行了局部的排序，ReduceTask只需保证Copy的数据的最终整体有效性即可。 

1- map端为什么需要将溢写得到的多个小的磁盘文件合并为最终的一个大的磁盘文件？

答：

​ 1- 小文件和大的文件存储的内容都是一样，但是多个小文件占用的存储空间比大文件要多一些

​ 2- （核心原因）文件在操作的时候，底层需要进行磁盘IO（https://zhuanlan.zhihu.com/p/443657373）。如果小文件过多，磁盘IO需要进行比较长的时间。

2- 为什么map端将数据写入磁盘的时候，还需要对数据进行sort，也就是局部排序？

答：因为reduce需要去map端读取数据。如果对数据排序了，那么reduce读取数据的速度相对比没排序的要快。

2.3 reduce阶段流程

第一阶段是Reducer任务会主动从Mapper任务复制其输出的键值对。Mapper任务可能会有很多，因此Reducer会复制多个Mapper的输出。第二阶段是把复制到Reducer本地数据，全部进行合并，即把分散的数据合并成一个大的数据。再对合并后的数据排序。该排序是在reduce端进行的，进行排序是为了提升reduce对数据的处理速度。第三阶段是对排序后的键值对调用reduce方法。键相等的键值对调用一次reduce方法，每次调用会产生零个或者多个键值对。最后把这些输出的键值对写入到HDFS文件中。