1 Split 阶段

在 mapreduce 中的 map task 开始之前，将文件按照指定的大小切割成若干个部分，每一部分称为一个 split，默认是 split 的大小与 block 的大小相等，均为 128MB。split 大小由 minSize、maxSize、blocksize 决定，以 wordcount 代码为例，以下是 main()方法

进入 waitForCompletion(true)方法，进入 submit()方法：

进行提交：

找到  return submitter .submitJobInternal(Job.this, cluster); 进入，找到  int maps = writeSplits(job, submitJobDir);

找到写分片机制，writeSplites,接着进入 进入 writeNewSplits() 方法；

进入 writeNewSplits()方法，可以看出该方法首先获取 splits 数组信息后，排序，将会优先处理大文件。最终返回 mapper 数量。

这其中又分为两部分：确定切片数量 和 写入切片信息。确定切片数量的任务交由 FileInputFormat 的 getSplits(job)完成，写入切片信息的任务交由 JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array)方法，该方法会将切片信息和 SplitMetaInfo 都写入 HDFS 中。

return array.length; 返回的是 map 任务数，

默认 map 的数量是：  default_num = total_size / block_size; 实际的 mapper 数量就是输入切片的数量，而切片的数量又由使用的输入格式决定，默认为 TextInputFormat，该类为 FileInputFormat 的子类。

确定切片数量的任务交由 FileInputFormat 的 getSplits(job)完成。FileInputFormat 继承自抽象类 InputFormat，该类定义了 MapReduce 作业的输入规范，其中的抽象方法 List getSplits(JobContext context)定义了如何将输入分割为 InputSplit，不同的输入有不同的分隔逻辑，而分隔得到的每个 InputSplit 交由不同的 mapper 处理，因此该方法的返回值确定了 mapper 的数量。

2 Map 阶段

每个 map task 都有一个内存缓冲区， map 的输出结果先写到内存中的环形缓冲区，缓冲区为 100M,不断的向缓冲区力写数据，当达到 80M 时，需要将缓冲区中的数据以一个临时文件的方式存到磁盘，当整个 map task 结束后再对磁盘中这个 map task 所产生的所有临时文件做合并，生成最终的输出文件。

最后，等待 reduce task 来拉取数据。当然，如果 map task 的结果不大，能够完全存储到内存缓冲区，且未达到内存缓冲区的阀值，那么就不会有写临时文件到磁盘的操作，也不会有后面的合并。在写入的过程中会进行分区、排序、combine 操作。

环形缓冲区：是使用指针机制把内存中的地址首尾相接形成一个存储中间数据的缓存区域，默认 100MB;80M 阈值,20M 缓冲区,是为了解决写入环形缓冲区数据的速度大于写出到 spill 文件的速度是数据的不丢失;Spill 文件：spill 文件是环形缓冲区到达阈值后写入到磁盘的单个文件.这些文件在 map 阶段计算结束时,会合成分好区的一个 merge 文件供给给 reduce 任务抓取;spill 文件过小的时候,就不会浪费 io 资源合并 merge;默认情况下 3 个以下 spill 文件不合并;对于在环形缓冲区中的数据,最终达不到 80m 但是数据已经计算完毕的情况,map 任务将会调用 flush 将缓冲区中的数据强行写出 spill 文件。

按照 key 合并成大文件，减少网络开销

2.1 分区

假设有听个 reduce 任务，则分区的计算如下：

2.2 排序

在对 map 结果进行分区之后，对于落在相同的分区中的键值对，要进行排序。

3 Shuffle 阶段

Shuffle 过程是 MapReduce 的核心，描述着数据从 map task 输出到 reduce task 输入的这段过程。

reducetask 根据自己的分区号，去各个 maptask 分区机器上取相应的结果分区数据，reducetask 会将这些文件再进行合并（归并排序)。

所有相同 key 的数据汇集到一个 partition。

将相同的 key value 汇聚到一起， 但不计算

4 Reduce 阶段

reduce 阶段分三个步骤：

1 reduce 任务会创建并行的抓取线程（fetcher）负责从完成的 map 任务中获取结果文件，是否完成是通过 rpc 心跳监听，通过 http 协议抓取；默认是 5 个抓取线程，可调，为了是整体并行，在 map 任务量大，分区多的时候，抓取线程调大；

2 抓取过来的数据会先保存在内存中，如果内存过大也溢出，不可见，不可调，但是单位是每个 merge 文件，不会切分数据；每个 merge 文件都会被封装成一个 segment 的对象，这个对象控制着这个 merge 文件的读取记录操作，有两种情况出现：在内存中有 merge 数据 •在溢写之后存到磁盘上的数据 •通过构造函数的区分，来分别创建对应的 segment 对象

3 这种 segment 对象会放到一个内存队列中 MergerQueue，对内存和磁盘上的数据分别进行合并，内存中的 merge 对应的 segment 直接合并，磁盘中的合并与一个叫做合并因子的 factor 有关（默认是 10）

排序问题，MergerQueue 继承轮换排序的接口，每一个 segment 是排好序的，而且按照 key 的值大小逻辑（和真的大小没关系）；每一个 segment 的第一个 key 都是逻辑最小，而所有的 segment 的排序是按照第一个 key 大小排序的，最小的在前面，这种逻辑总能保证第一个 segment 的第一个 key 值是所有 key 的逻辑最小文件合并之后，最终交给 reduce 函数计算的，是 MergeQueue 队列，每次计算的提取数据逻辑都是提取第一个 segment 的第一个 key 和 value 数据，一旦 segment 被调用了提取 key 的方法，MergeQueue 队列将会整体重新按照最小 key 对 segment 排序，最终形成整体有序的计算结果；

Reduce 任务数量 在大数据量的情况下，如果只设置 1 个 Reduce 任务，其他节点将被闲置，效率底下 所以将 Reduce 设置成一个较大的值（max：72）.调节 Reduce 任务数量的方法 一个节点的 Reduce 任务数并不像 Map 任务数那样受多个因素制约

参考博文：

https://gitchat.blog.csdn.net/article/details/104190423?spm=1001.2101.3001.6650.7&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EOPENSEARCH%7ERate-7-104190423-blog-81115646.pc_relevant_multi_platform_whitelistv4&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EOPENSEARCH%7ERate-7-104190423-blog-81115646.pc_relevant_multi_platform_whitelistv4&utm_relevant_index=8