MapReduce框架原理

一、MapReduce框架原理

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

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

1.数据切片与MapTask并行度决定机制

MapTask并行度由切片个数决定，切片个数由输入文件和切片规则决定

1. 一个job的Map阶段并行度由客户端在提交job时的切片数决定
2. 每一个Split切片分配一个MapTask并行实例处理
3. 默认情况下，切片大小=BlockSize（块的大小）
4. 切片时不考虑数据集整体，而是逐个针对每一个文件单独切片
5. 提交切片规划文件到YARN上，YARN上的MrAppMaster就可以根据切片的规则计算开启MapTask个数。（按照切片的个数来开启MapTask的个数，切了多少片，开启多个MapTask。）

2.Job提交流程源码

1. waitForCompletion()
2. submit();
3. 建立连接：判断是在本地运行环境还是yran集群运行环境
4. 提交job：
   ①创建给集群提交数据的stag路径
   ②创建job，获取job的id放到stag路径
   ③如果是集群模式就把对应的jar包拷贝到相对于的路径，不是就不拷贝
   ④计算切片，生产切片规划文件
   ⑤向stag路径写XML配置文件
   ⑥提交job，返回提交的状态

3.FileInputFormat切片源码解析

1. 程序先找到你数据存储的目录
2. 开始遍历处理（规划切片）目录下的每一个文件
3. 遍历每一个文件（每一个文件单独切片）
   ①获取文件大小
   ②计算切片的大小=BlockSize块的大小，默认是128M
   ③开始切片从0-128M为第一片，以此类推（每次切片都会判断剩下文件大小是否大于当前切片大小的1.1倍，大于正常切片，小于不再切片）
4. 提交切片规划文件到Yarn上，Yarn按照切片规划文件开启MapTask个数。

4.TextInputFormat和CombineTextInputFormat

在运行MapReduce程序时，输入的文件格式包括：基于行的日志文件、二进制格式文件、数据库表等。那么，针对不同的数据类型，用不同接口实现类进行处理。

FileInputFormat常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。

框架默认的TextInputFormat切片机制是对任务按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个MapTask，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。

CombineTextInputFormat用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个MapTask处理。

5.MapReduce详细工作流程☆☆☆

1. 准备一个待处理文件 比如：文件大小200M
2. 在客户端submit()前，获取待处理数据信息，根据参数配置形成一个切片规划文件。（切片的原理就上面的源码解析）如：0-128第一片，128-200是第二片
3. 提交信息给Yarn，信息包括切片信息，jar包，xml配置文件
4. Yarn开启MrAppmaster（整个集群运行的老大）读取切片信息，根据切片的个数 从而开启MapTask数量
5. MapTask启动之后开读取文件中的数据，默认使用TextInputFormat中的RecorderReader是按行进行读取，其中key对应的偏移量，value对应的一行数据
6. Map接收读取的KV，根据业务需求编写代码
7. Map方法之后，数据首先进入到分区方法，对数据进行分区，然后向环形缓冲区写入数据，默认大小为100M，当环形缓冲区到达80%的时候，开始进行溢写，溢写前对数据进行排序，排序是按照对key的索引进行字典顺序排序，排序使用的是快排；溢写产生大量溢写文件，需要对溢写文件进行归并排序；对溢写的文件也可以进行Combiner操作，前提是汇总操作，求平均值不行。最后将文件按照分区存储到磁盘，等待Reduce端主动拉取数据。
8. 每个Reduce拉取Map端对应分区的数据。拉取数据后先存储到内存中，内存不够了，再存储到磁盘。拉取完所有数据后，采用归并排序将内存和磁盘中的数据都进行排序。在进入Reduce方法前，可以对数据进行分组操作。
9. Reduce根据业务需求对传入的数据进行汇总
10. OutputFormat将最终的结果写入到HDFS

第7步和第8步的过程叫Shuffle，也就是说Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle

第7步和第8步也可以简单说：

• Partition：按照一定的规则对key，value的list进行分区
• sort：对每个分区内的数据进行排序
• Combiner：在Map端进行举报聚合汇总。目的是为了减少网络带宽的开销，提高效率
• 然后再进行分组操作

6.分区说明

Partition分区的规则：对map输出的key取哈希值，用这个哈希值与reducetask的个数值取余。余几就将这个key，value放在对应的分区编号里

7.排序说明

MapTask和ReduceTask均会对数据按照key进行排序，该操作属于hadoop默认行为，默认是按照字典顺序排序，实现该排序的方法是快排

8.Combiner合并

9.ReduceTask并行度决定机制

（1）设置ReduceTask并行度（个数）
ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，ReduceTask数量的决定是可以直接手动设置：

// 默认值是1，手动设置为4
job.setNumReduceTasks(4);

注意：ReduceTask设置的个数需要测试的多个比较合适

注意事项：