大数据|MapReduce工作机制大数据|hadoop

MapReduce工作流程
【大数据|MapReduce工作机制】
文章图片

1、InputFormat根据job提供的文件目录，对目录下面的文件进行遍历 2、首先遍历第一个文件

获取文件的大小
获取切片大小

文章图片

上面的源码就是切片大小的计算公式，blocksize为块的大小；如果job运行在本地，默认块大小是32M；如果job运行在yarn上，hadoop1.x 默认块大小是64M；hadoop2.x 默认块大小是128M。
minSize=1 maxSize=long.max_value
先对块大小和maxSize取最小值，最小值为块大小；再对块大小和minSize取最大值，得到最大值为块大小；所以一般情况下，切片大小为块的大小128M。
如果想要增加切片大小，可以设置minSize>128M
如果想要减小切片大小，可以设置maxSize<128M
对文件进行切片，假设第一个文件300M
第一个切片:0-128M
第二个切片:128-256M
第三个切片:256-300M
1、每次切完片都会对剩下文件的大小进行判断，如果剩余文件大小没有超过块大小的1.1倍，超过1.1继续切片，没有超过则将剩余的文件当成一个切片。
举例：一个文件260M，第一个切片0-128，剩余132M，132<128*1.1，所以剩下的132M被分为一个切片；该文件一共2个切片，第一个128M，第二个132M
2、在进行切片的时候，如果一个单词在一行中没有写完，那么切片的时候，就会跳到下一行，把该单词读取完整，不会出现把一个单词分成两个部分的情况。

3、所有文件的切片都是在getSplit()方法中完成的，方法返回切片的信息数组inputSplit[], inputSplit只是记录了切片的元数据，包括切片的长度，所在节点等 4、切片信息提交到HDFS上，AplicationMaster根据切片信息来决定启动几个MapTask 5、MapTask启动之后，通过InputFormat来读取文件(默认是TextInputFormat)，一次读取一行，k为行的偏移量，v为一行的内容 6、k,v进入到Mapper中的Map方法,对k，v进行数据处理，每一行数据都会进入到这个方法 7、Context.write将新的k，v写入到OutputCollector，OutputCollector再把数据输出到环形缓冲区 8、环形缓冲区默认大小是100M，一边是K,V数据，另一边是数据的元信息(key的起始位置，v的起始位置，分区，v的长度) 9、当数据达到80%之后，就会进行溢写，将数据写入到磁盘中；在溢写之前会对数据进行分区和排序，排序的依据是字典排序，方法是快排；一次溢写完成后，会清除内存中的内容，方便后续数据写入

分区数量可以通过设置reducetask的数量来决定，有几个reducetask就有几个分区
系统默认按照map输出key 的hashcode进行排序

文章图片

Key的hashcode按位与Integer的最大值，保证hash值不为负数；再对ReduceTasks的数量取余，numReduceTasks默认是1，所以我们不设置的话，默认是一个分区；
当我们设置多个分区的时候，按照hashcode来进行分区，就有可能产生数据偏移现象；一个分区的数据很多，另一个分区的数据很少，导致reduce一个很忙，一个很闲。这个时候我们就可以通过对数据加盐或者再hash的方式，对数据进行打散，减少出现数据偏移的情况

10、在溢写的时候，可以加入combiner，进行一次汇总，减少网络传输；combiner和reduce类似，但是combiner只针对节点上的MapTask，reduce是对所有的MapTask 11、文件的大小不同可能会产生多次溢写，在全部溢写完成后，会进行一次归并排序，有几个分区，就有几个文件，且区内有序。溢写到磁盘中的文件，最后会自动删除 12、Reducetask根据自己的分区号，去每个MapTask节点上拷贝所属分区的数据，数据一开始放在内存中，当超过一定的大小，就会写入到磁盘中 13、所属分区的文件全部拷贝完成后，会进行一次归并排序，将所有小文件生成一个大文件 14、对大文件中的数据进行分组，形成(k,values) 假设一开始为(hive,1),(hive,1),分组后就变成了(hive,2)
15、进入reduce程序，对分组后的数据进行逻辑处理，每组相同的key只会进入一次reduce 16、最后Reduce得到的结果通过OutputFormat输出到文件中