MapReduce流程

Posted 2021-01-29 sgs-sunguishi

一、InputSplit

• MapReduce框架调用job.setINputFormatClass定义的InputFormat读取数据
• InputFormat接口里包括两个方法：getSplits()和createRecordReader()，这两个方法分别用来定义输入分片和读取分片的方法。

　　1、计算分片

• 调用InputFormat的 getSplits()负责将文件切分成多个分片(InputSplit)，但InputSplit并没有实际切分文件，而只是说明了如何切分数据（一个分片长度和一个记录数据的位置的数组），InputSplit只是逻辑上的切分。
• InputSplit类定义了如何切分文件，它包括了getLength()和getLocation()两个方法。getLength()方法用来获取InputSplit的大小，getLocation()则用来获取InputSplit的位置。
• 计算最大分片大小splitSize的方法是：Math.max(minSize, Math.min(maxSize, blockSize))。
• 每个InputSplit对应一个Mapper任务。
• MapReduce有一个机制：对于每一个文件判断其切完后剩余长度的长度是否>128*1.1,如果是，就分成一个切片，如果不是就把剩余的切片当做一个切片。
• Hadoop是为了处理大文件的。通过CombineFileSplit把很多小文件包在一个InputSplit内就可以用一个MapTask处理很多小文件。

　　2、读取分片

• InputFormat的RecordReader，用来从InputSplit读取记录，提供map任务所要处理的key-value对。
• 在InputFormat的默认子类TextInputFormat中，提供了LineRecordReader。 LineRecordReader会把文件的每一行作为一个单独的记录，并以行偏移为键值。这也就解释了WordCount例子中，行偏移为key值，每一行的内容作为value的原因。

 

二、Mapper

调用Job的setMapperClass指定自定义的Mapper的map()方法，将RecordReder解析的每个键/值<LongWritable, Text>对输入给Map的map()方法

最终是按照自定义的Map的输出<key，value>

 

三、Shuffle（混排）

• 将map的输出作为reduce的输入的过程就是shuffle
• Map会先将数据输出到环形缓存区保存，大小默认为100M，环形缓存区的数据达到一个特定的阀值（默认80%）时，系统将会启动一个后台线程把缓冲区中的内容spill到磁盘，生成一个spill文件。Map的输出将会继续写入到剩余的20%缓冲区，如果缓存区被撑满了，Map就会被阻塞直到spil完成。
• 环形缓冲区的数据写入磁盘前,程序会对他进行一个二次排序，首先（combiner可选）然后根据数据所属的partition排序，最后每个partition中再按Key排序。
• Map结束后，会将多个spill的文件归并排序合并（combiner可选）成一个大文件，以供不同的reduce来拉取相应的数据。
• 如果 reducer的数量设为0，那么map的输出会直接由输出组件，输出到输出目录指定的文件中。

• Combiner就是一个Mini Reducer，它在执行Map任务的节点本身运行，先对Map的输出作一次简单的Reduce，使得Map的输出更紧凑，更少的数据会被写入磁盘和通过网络传送到Reducer。

• partitioner的作用是将mapper/combiner输出的键/值对拆分为分片（shard），每个reducer对应一个分片。
• 调用Job的setPartitionerClass指定自定义的partitioner：
• 定义一个类，并且继承Partitioner类，重写其getPartition方法；该方法包含三个参数：输入是Map的结果对<key, value>和reducer个数， 均由框架自动传入，输出则是分片编号。默认实现key.hashCode()%(reducer的个数)

 

四、ReduceTask

• 在reduce阶段，reducer首先下载多个映射到这个reducer的map输出
• 下载完后调用Job的job.setSortComparatorClass设置的key比较函数类对所有数据排序（二次排序）。然后开始构造一个key对应的value迭代器。
• 排序的目的是将相同键的记录聚合在一起，这样其所对应的值就可以很方便地在reduce任务中进行迭代处理。这个过程完全不可定制，而且是由框架自动处理的。开发人员只能通过自定义Comparator对象来确定键如何排序和分组。
• 分组，使用job.setGroupingComparatorClass设置的分组函数类。只要这个比较器比较的两个key相同，他们就属于同一个组（它们的value放在一个value迭代器，而这个迭代器的key使用属于同一个组的所有key的第一个key）
• 最后就是进入Reducer的reduce方法，reduce方法的输入是所有的key和它的value迭代器。执行reduce函数。
• OutPutFormat获取reduce函数输出<key,value>，并通过RecordWrite将结果写入到OutputFormat中准备好的位置。
• OutPutFormat的checkOutputSpecs主要检查输出目录是否合法，一般在作业提交之前会被调用。

 

 

学习阶段 如有错误 恳请指正！！！