MapReduce核心编程思想和原理(图形化通俗易懂)

MapReduce核心编程思想和原理

• • 组件模块说明
  • MapReduce核心编程思想
  • MapReduce 框架原理
  • MapReduce 工作流程
  • Shuffle机制

组件模块说明

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

MapReduce 进程：一个完整的 MapReduce 程序在分布式运行时有三类实例进程，分别为MrAppMaster，MapTask，ReduceTask。

MrAppMaster：负责整个程序的过程调度及状态协调。

MapTask：负责 Map 阶段的整个数据处理流程。并行处理输入数据。

ReduceTask：负责 Reduce 阶段的整个数据处理流程。对 Map 结果进行汇总。

数据块： Block 是 HDFS 物理上把数据分成一块一块。 数据块是 HDFS 存储数据单位。

数据切片： 数据切片只是在逻辑上对输入进行分片， 并不会在磁盘上将其切分成片进行存储。数据切片是 MapReduce 程序计算输入数据的单位，一个切片会对应启动一个 MapTask。

Combiner：是MR程序中Mapper和Reducer之外的一种组件。Combiner是在每一个MapTask所在的节点运行。Combiner的作用为对每一个MapTask的输出进行局部汇总，以减小网络传输量。

MapReduce核心编程思想

图中分为2个文件，一个200M，一个100M。一个block大小默认为128M，则第一个文件128M分配给一个MapTask，剩下72M分配给另外一个MapTask。

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

MapReduce 框架原理

InputFormat：

1. 可以对输入进行控制，可以用FileInputFormat、TextInputFormat、CombineTextInputFormat等。
2. 对输入的文件进行数据切片(默认情况下切片大小=blocksize，切片时不考虑数据集整体，而是逐个针对每一个文件单独切片)，形成多个InputSplit文件，每一个InputSplit对应着一个map任务。

Mapper：数据源通过InputFormat取读后，交给Mapper进行后续业务逻辑(用户自己写的代码)处理。

Shuffle：包含排序、分区、压缩、合并等等。

Reducer：拉取Mapper阶段处理的数据，拉的过程中，要经历shuffle的过程。

OutputFormat：对输出的控制，比如可以输出到文件、mysql、Hbase、ES等。

MapReduce 工作流程

1. 待处理文件200m。
2. 客户端submit()前，获取待处理数据的信息，然后根据参数配置，形成一个任务分配的规划。默认按128M切片，分为 0~128和 128~200。
3. 客户端提交到集群包含：Job.split(job的切片)，wc.jar(代码)，Job.xml(job运行相关的参数)。
4. YARN的ResourceManager(整个集群所有资源的管理者)开启Mrappmaster(单个任务运行的老大，为应用程序申请资源并分配给内部的任务)，Mrappmaster会取读Job.split切片信息，根据切片个数(图中2个)开启MapTask个数。
5. MapTask启动后，通过InputFormat(默认实现是TextInputFormat，方法是createRecordReader，按行读LineRecordReader)读输入的文件。
6. 数据源通过InputFormat取读后，交给Mapper进行后续业务逻辑运算(用户自己写的代码)处理。
7. outputCollector输出收集器，向环形缓冲区写入数据，其实就是一块内存，一半用于存数据(key;value)，另外一半存索引(描述数据的元数据，index为索引；partition为分区；keystart指key在内存存储在什么位置；valstart指value在内存存储在什么位置)。outputCollector默认大小100M。当写入80%的数据后(为什么80%是因为可以边接收数据边往磁盘溢写数据)，开始反向写，把数据溢写到磁盘。
8. 在溢写之前会将缓冲区的数据按照指定的分区规则（默认分区是根据key的hashCode对ReduceTasks个数取模得到的，用户没法控制哪个key存储到哪个分区。但是可以自定义）进行分区和排序。图中2个分区，分区1会进入reduce1，分区2会进入reduce2，互相不影响。排序是对分区内的数据进行排序，对index(索引)通过快排按字典顺序进行排序。
9. 当写入80%的数据后(或者数据已经全部处理完)，就会把环形缓冲区的数据溢写到磁盘。可能发生多次溢写，溢写到多个文件。
10. 对所有溢写到磁盘的文件(已经有序，可以通过归并来排)进行归并排序合成一个文件。保证每个分区的数据是有序的。

承接上一张图

11. Combine合并，预聚合(优化手段)，可以对每个MapTask的输出进行局部汇总，以减少网络传输量。 12. MrappMaster，所有MapTask任务完成后，启动相应数量的ReduceTask，并告知ReduceTask处理数据范围（数据分区）。 13. ReduceTask主动从MapTask对应的分区，拉取数据。因为虽然每个MapTask的数据已经是有序，但是会从多个MapTask拉取数据，所以还要进行归并排序。 14. 将数据传给reduce进行处理，一次读取一组数据。 15. GroupingComparator，用的比较少。hadoop默认分组是按key，也就是一个key是一组，GroupingComparator主要的作用是可以决定哪些数据作为一组。 16. 最后通过OutputFormat输出，默认是TextOutputFormat。

Shuffle机制

Map 方法之后， Reduce 方法之前的数据处理过程称之为 Shuffle。

首先，通过getPartition获取是哪个分区。标记分区后，进入环形缓冲区。一半用于存数据，另外一半存索引。当写入80%的数据后，就会反向溢写。在溢写之前会将缓冲区的数据进行排序。之后可以进行Combiner(可选)。然后进行多次溢写，一个是spill.index(索引)，一个是Spill.out(数据)。之后对所有溢写到磁盘的文件进行归并排序。之后可以进行Combiner(可选)。之后可以设置压缩(提高传输效率)。之后数据写到磁盘上，等待reduce拉取数据。

ReduceTask主动从MapTask对应的分区，拉取数据。先尝试把数据存在内存里。如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后，ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。然后做分组(按相同key分组)。再进入Reduce方法。