MapReduce最核心的部分是它的MapTask阶段和ReduceTask阶段。可以简单的理解MapTask阶段用于将一行数据拆分成一个KV的数据,ReduceTask阶段是对Map阶段生成的KV数据做统计。
Hadoop的MapReduce的大致流程。
- 按照块对原始数据进行切片。比如一个文件200M,Hadoop会先将其在逻辑上分成128M+72M,这样就有两个块数据
- 每个块数据都会由一个MapTask来处理,MapTask最终生成分区数据。
- ReduceTask对所有MapTask的数据进行聚合
一、MapTask工作机制
MapTask主要分为6个阶段:Read阶段、Map阶段、Collect阶段、Spill阶段、Merge阶段
- Read阶段:MapTask通过InputFormat获取的RecordReader,从输入流中解析出K/V(Key/Value)数据,对于一个文件而言,K就是行号,V就是一行内容。
- Map阶段:该阶段属于用户自定义阶段(对应于Mapper类),接收Read阶段解析出的KV数据,解析并生成新的KV数据。
- Collect阶段:在Map阶段最后都需要将新生成的KV数据通过context.write()写出去,最终都会通过OutputCollector.collect()输出结果。在该函数内部还会调用Partitioner进行分区,最终会将数据写入到环形缓冲区。
- Spill阶段:也称“溢写”阶段,在Collect阶段会不断向环形缓冲区写入数据,环形缓冲的数据是存放在内存中的,当数据过多时(一般会将内存等分成2部分,写数据时先向一遍写入,当数据超过80%时,认为该部分数据已满,后续写入的数据会写道另一部分),需要将数据从内存写出到磁盘,在写磁盘之前,会对内存的中的数据做排序,并在必要时对数据进行合并(Combiner)、压缩。最终的数据会写到一个临时文件中。
- Merge阶段:对Spiil阶段产出的临时文件进行合并,最终生成一个数据文件。
对于Spill阶段做详细的说明,方便理解。
- ?利用快速排序对数据进行排序,排序的方式是,先按照分区进行排序,再按照Key进行排序,最后得到的数据以分区单位聚集在一起,且同一分区内数据按照Key有序。
- 按照分区编号将数据写入到临时文件output/spillN.out(N表示当前溢写次数)中,如果设置了Combiner,则在写入之前会做以此数据聚集。
- 将分区数据的元信息写道内存索引数据结构(SpillRecord)中,其中每个分区的元信息包括:在临时文件中的偏移量、压缩前的数据大小、压缩后的数据大小。如果当前内存索引大小超过1MB,则将内存索引写到文件output/spillN.out(N表示当前溢写次数)中。
二、ReduceTask工作机制?
ReduceTask主要分为3个阶段:Copy阶段、Sort阶段、Reduce阶段。
- Copy阶段:ReduceTask从MapTask上拷贝一片数据,并针对某一片数据,如果其大小超过一定阈值,则写到磁盘上,否则直接放到内存中。
- Sort阶段:在远程拷贝数据的同时,ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并,以防止内存使用过多或磁盘上文件过多。按照MapReduce语义,用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起,Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序,因此,ReduceTask只需对所有数据进行一次归并排序即可。
-
Reduce阶段:reduce()函数将计算结果写到HDFS上。
?三、ReduceTask并行决定机制
MapTask的数量可以根据切片的数量来决定。但是ReduceTask是可以手动设置的。
// 同时指定相应数量的ReduceTask
job.setNumReduceTasks(5);
关于Reduce数量的设置有一些注意事项:
- ReduceTask=0,说明没有Reduce阶段,输出文件的个数与Map个数相同。
- ReduceTask默认为1,所以输出一个文件。
- 如果数据分布不均匀,会在Reduce阶段产生数据倾斜。
- ReduceTask的数量不是随意设置的,需要根据业务来设置,如果需要全局的汇总统计,那么只能有一个ReduceTask。
- 具体有多少ReduceTask由集群性能决定。
- 如果分区数>1,但是ReduceTask=1,那么不会再进行分区,只有ReduceTask的数量>1时才会进行分区。