本文共 6046 字，大约阅读时间需要 20 分钟。

一.MapReduce工作流程图片如下

二.工作流程机制详解

1.TextInputFormat读取文件详细解析

TextInputFormat的源码注释为:

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

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

• 1.验证作业的输入规格
• 2.将输入文件分成逻辑块(默认大小等于block的大小(通常为128M))，然后将每个逻辑文件分配给一个单独的Mapper
• 3.提供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;//获取RecordReaderRecordReader
   
     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
   
     splits = new ArrayList
    
     (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负责处理这个视图并呈现一个面向记录的视图

因此我们知道这里切片只是一个逻辑划分,记录了一些切片信息

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端

注意:

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

转载地址：http://fcjxi.baihongyu.com/