文章目录

一.MapReduce工作流程图片如下二.工作流程机制详解1.TextInputFormat读取文件详细解析2.map端(注意map task 数量由切片数量决定)3.reduce端

一.MapReduce工作流程图片如下

二.工作流程机制详解

1.TextInputFormat读取文件详细解析

TextInputFormat的源码注释为:

用于纯文本文件,文件会被分成几行,使用换行或回车来表示行尾.键是文件中的位置(每行的偏移量,会进行累加)，值是文本行。其中有一个方法为返回一个LineRecordReader<LongWritable, Text> 字符编码以utf8格式传输 我们发现TextInputFormat这个类继承自FileInputFormat，而FileInputFormat这个类实现了InputFormat接口

查看InputFormat接口的源码注释我们了解到这个接口的作用为:

1.验证作业的输入规格2.将输入文件分成逻辑块(默认大小等于block的大小(通常为128M))，然后将每个逻辑文件分配给一个单独的Mapper3.提供RecordReader接口,用于从逻辑块中收集输入记录，以供Mapper处理4.通常是FileInputFormat子类(TextInputFormat)根据输入的总大小（以字节为单位）将输入拆分为逻辑块5.输入文件的block块大小被视为输入拆分的上限,分割尺寸的下限可以通过设置配置文件来改变(mapred-default.xml文件的mapreduce.input.fileinputformat.split.minsize属性)6.由于要遵守记录边界(物理分块是不管记录边界的),因此对于许多应用程序,基于输入大小的逻辑拆分是不够的,在这种情况下,应用程序还必须实现一个RecordReader类- - 它负责记录边界(一般是各种文件的换行符),并向单个任务提供逻辑切片的记录。

在InputFormat的源代码中有如下两个方法：

//获取切片(一个切片就是一个Mapper任务)数组 InputSplit[] getSplits(JobConf job, int numSplits) throws IOException; //获取RecordReader RecordReader<K, V> getRecordReader(InputSplit split, JobConf job, Reporter reporter) throws IOException;

追踪FileInputFormat接口,切片方法如下:

/** Splits files returned by {@link #listStatus(JobConf)} when they're too big.*/ public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException { //开启线程 Stopwatch sw = new Stopwatch().start(); //获取文件状态 FileStatus[] files = listStatus(job); // 为了度量所有文件总长度,需要设置文件输入数和每个文件长度(job传的可能是一个目录,目录下有很多文件) job.setLong(NUM_INPUT_FILES, files.length); long totalSize = 0; // 计算文件总大小 for (FileStatus file: files) { if (file.isDirectory()) { // 检查有效文件 throw new IOException("Not a file: "+ file.getPath()); } totalSize += file.getLen(); } //默认切片是block大小(128M)如果自己定义了切片的大小就按照设置参数来,所有文件大小/切片数-->就可以求得目标切片大小 long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits); //最小的切片大小只要不动配置文件参数默认是1,如果按设置了就按设置参数来 long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize); // 生成切片步骤 //准备一个文件切片的集合-->需要去看看文件切片的机制 ArrayList<FileSplit> splits = new ArrayList<FileSplit>(numSplits); NetworkTopology clusterMap = new NetworkTopology();//网络拓扑,这个先不管 for (FileStatus file: files) { //文件路径(namenode会存储这些信息) Path path = file.getPath(); //文件长度(namenode会存储这些信息) long length = file.getLen(); if (length != 0) { FileSystem fs = path.getFileSystem(job); //获取文件的block块(只会获取一次完整文件的块(找到一个活的块就不会找他的备份)) BlockLocation[] blkLocations; if (file instanceof LocatedFileStatus) { blkLocations = ((LocatedFileStatus) file).getBlockLocations(); } else { blkLocations = fs.getFileBlockLocations(file, 0, length); } if (isSplitable(fs, path)) { //获取block块大小(默认128M) long blockSize = file.getBlockSize(); //计算切片大小:在(目标大小,最小大小(设置的参数),block块大小)中取中值的操作 long splitSize = computeSplitSize(goalSize, minSize, blockSize); long bytesRemaining = length; //SPLIT_SLOP指的是切片的溢出系数为1.1 //（每次切片时，都要判断切完剩下的部分是否大于块的 1.1倍，不大于 1.1 倍就划分一块切片）。 while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) { String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length-bytesRemaining, splitSize, clusterMap); //记录了路径,切片偏移量,切片大小等,加入文件切片的集合中 splits.add(makeSplit(path, length-bytesRemaining, splitSize, splitHosts[0], splitHosts[1])); //切完一次剩余的切片大小 bytesRemaining -= splitSize; } if (bytesRemaining != 0) { String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length - bytesRemaining, bytesRemaining, clusterMap); splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining, splitHosts[0], splitHosts[1])); } } else { String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,0,length,clusterMap); splits.add(makeSplit(path, 0, length, splitHosts[0], splitHosts[1])); } } else { //Create empty hosts array for zero length files splits.add(makeSplit(path, 0, length, new String[0])); } } sw.stop(); if (LOG.isDebugEnabled()) { LOG.debug("Total # of splits generated by getSplits: " + splits.size() + ", TimeTaken: " + sw.elapsedMillis()); } //将集合转为数组 return splits.toArray(new FileSplit[splits.size()]); }

追踪RecordReader接口,发现其功能为:

1.从一个逻辑块中读取键值对2.它转换了从输入端的面向字节的视图到面向记录边界的视图3.因此，它承担处理记录边界并为任务提供键和值的责任。

追踪InputSplit抽象类,查看注释发现其功能为:

1.一个切片代表了一个一个Mapper任务2.通常，它在输入上呈现一个面向字节的视图,RecordReader负责处理这个视图并呈现一个面向记录的视图

因此我们知道这里切片只是一个逻辑划分,记录了一些切片信息 数据切片只是在逻辑上对输入数据进行分片，并不会在磁盘上将其切分成分片进行存储。 InputSplit 只记录了分片的元数据信息，比如起始位置、长度 以及所在的节点列表等 提交切片规划文件到Yarn上，Yarn上的 MrAppMaster 就可以根据切片规划文件计算开启 map task 个数 在map task的read阶段接受输入分片,根据RecordReader按照InputSplit 记录 的位置信息读取数据,会将一个分片处理成一行一行的数据即一个键值对键为行偏移量,值为文本数据

2.map端(注意map task 数量由切片数量决定)

1.我们知道map端的输入是上步的一个个键值对,会对每个键值对执行一次map方法,在map方法里根据业务需求转换成一个新的键值对用context.write()方法写到OutputCollector中注意:输入输出的键值对的类型必须都可序列化2.再由OutputCollector将收集到的键值对写入到环形缓冲区中(默认100M)1)因为环形缓冲区达到80%会自动清理即溢写,不会造成数组下标越界情况2)环形缓冲区可使得内存使用最优,性能较好 'shuffle是MapReduce处理流程中的一个过程,他的每一个处理步骤是分散在各个map任务和reduce任务上完成的.整体看,有三个操作' 1)分区 2)Sort根据key排序 3)Combiner进行局部合并 3.需要注意的是溢写之前需要对数据分区排序,对环形缓冲区里面的每个键值对hash一个分区数,hash后分区相同值在同一个分区,然后根据分区值和键两个关键字来升序排序,同一个分区内按照键排序(这就要求键类型实现的是WritableComparable接口)4.将环形缓冲区排序后的内存数据不断写到本地磁盘中,可能会出现很多文件5.多个文件会被merge合并成大文件,合并采用的是归并排序,最终文件还是分区且区内有序(这一步就是combiner,不一定要有,但是没有会增加网络传输和本地io压力)

3.reduce端

注意: reduce task 的数量的决定是可以直接手动设置的并不一定和map task 数量相等,默认是1 如果分区数不是 1，但是 reducetask 为 1，是不执行分区操作的(执行分区之前会进行判断)

1.reduce task 根据自己的分区号,去各个map task 节点拷贝相同分区的数据到reduce task 本地磁盘工作目录2.同一分区来自不同map端的多个文件会被merge合并成大文件,合并采用的是归并排序,大文件按照键有序3.在合并成大文件后整个mapreduce的shuffle过程就结束了,后面进入到reduce逻辑运算过程4.首先调用GroupingComparator对大文件里面的数据进行分组,每次取出一组键值对调用reduce方法进行业务处理5.通过OutputFormat方法将结果数据写到part-r-00000的结果文件中去