map函数处理的<k1,v1>从HDFS文件中获取流程源码解析

tonygsw 2014-11-30

0 引子:

mapreduce在执行任务的时候,是如何将外部文件进行切分,并将这些文件转换成<k1,v1>键值对方式的

(还记得 map-reduce基本概念和wordcount解析 文章中提到的<k1,v1>概念吗?)

一些总结性的话:

a) recordreader + inputsplit是数据输入处理阶段非常重要的两个概念。

b) inputsplit: 对原始输入数据的封装,封装原始数据源,这个数据源可以是hdfs文件系统,也可是DB,或者数据流

c) recordreader: 把inputsplit的数据转换成<k1,v1>

d) inputsplit和hdfs的block之间的关系是什么?? 
 在hdfs看来,mapreduce就是hdfs的客户端,
mapreduce默认inputsplit大小和hdfs的一个block相等,
但是并不说
mapreduce的一个inputsplit就对应hdfs的一个block,
mapreduce是看不到hdfs体内的一个个的block的,  因此默认情况下,被处理目标文件有多少block,就会产生

多少inputsplit,也就会对应多少个map任务,即block和inputsplit只有这种对照关系。

为什么设计成一个hdfs block 对应一个 inputsplit:

核心就是数据本地化,在当前机器上计算的数据尽量不要传输

1个inputsplit>1个block时,比如1个inputsplit使用2个block时,hdfs的这两个block不一定都存储在同一个节点上,那么必然会产生网络传输,将需要的另一个block 传输到计算的这个节点上.

inputsplit<1个block时,那么一个block的数据中就必然有一部分没有被处理,必然就会被别的mapper处理,必然增大网络传输的概率.

看下图:

 
map函数处理的<k1,v1>从HDFS文件中获取流程源码解析
 

1 FileInputFormat代码简析, 回答如何将源文件逻辑分割

package org.apache.hadoop.mapreduce.lib.input


读取hdfs目录下的文件,后将文件便利,在执行getSplits方法时会发现,for循环内代码中,
每一个文件不论是1K还是1G
都会被处理封装成对应个数的inputsplit ,即一个文件对应一个或者多个split
而每一个split都会对应一个map进程
如果小文件太多,那么开启map进程太多 势必浪费资源,操作系统开启,执行,挂起一个进程是很消耗资源的

  /** 
   * Generate the list of files and make them into FileSplits.
   */ 
  public List<InputSplit> getSplits(JobContext job
                                    ) throws IOException {
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job)); 
    long maxSize = getMaxSplitSize(job);

    // generate splits
    List<InputSplit> splits = new ArrayList<InputSplit>();
	// hdfs//master:9000/hello 则files为hello   hdfs//master:9000/ 则files为目录下的所有文件
    List<FileStatus>files = listStatus(job); // 读取job中的输入路径,进入
 
    for (FileStatus file: files) {// 对目标路径下的文件/文件数 进行截取成 split操作, 每个文件至少对应一个切片,一个切片对应一个map
	//如果每个文件都是1M,那么这个文件就在切片时对应一个split 也就对应一个map操作,1T的文件就1T个map,这种计算要消费多大的资源,这就是小文件不适合hdfs的原因,
	//而如果64个1m文件合并后在处理,对应map就缩小了63倍,这样对程序性能影响有多少 可想而知
      Path path = file.getPath();// 获取文件路径
      FileSystem fs = path.getFileSystem(job.getConfiguration());// 获取hdfs操作
      long length = file.getLen();
      BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
      if ((length != 0) && isSplitable(job, path)) { // 当目标文件长度>0并且是可split时
        long blockSize = file.getBlockSize();// 文件块大小 默认是64M
        long splitSize = computeSplitSize(blockSize, minSize, maxSize);// 计算切片尺寸,进入此方法,可见结果为64M, 参数minSize值为1,maxSize为long的最大值
        如果想调整切片大小,需要重写方法 getMaxSplitSize    public static long getMinSplitSize( 两个方法
		
	
        long bytesRemaining = length;
        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
          splits.add(new FileSplit(path, length-bytesRemaining, splitSize, 
                                   blkLocations[blkIndex].getHosts()));
          bytesRemaining -= splitSize;
        }
        
        if (bytesRemaining != 0) {
          splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, 
                     blkLocations[blkLocations.length-1].getHosts()));
        }
      } else if (length != 0) { // 文件不可拆分
        splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
      } else { // 文件长度=0
        //Create empty hosts array for zero length files
        splits.add(new FileSplit(path, 0, length, new String[0]));
      }
    }
    
    // Save the number of input files in the job-conf
    job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());

    LOG.debug("Total # of splits: " + splits.size());
    return splits;
  }



  
  如下: 
  protected List<FileStatus> listStatus(
   ....
   Path[] dirs = getInputPaths(job); // 进入   大概看下 )



仅仅是对1T数据按照64M一个个的做标记, 是逻辑切分,不是物理切分,仅仅记录原始数据位置和要处理的长度
类比于高速公路上的 37公里处, 70公里处的路标, 仅仅是个标志而已,实际上高速公路还是连在一起的
看下面的数据结构,如果是物理切分的话,其数据结构必然也会要有 private bytes[] buf; 这个属性
public class FileSplit extends InputSplit implements Writable {
  private Path file;
  private long start;
  private long length;
  private String[] hosts;

 。。。。
}

2 RecordReader 代码简析,回答如何将split转换成键值对:

package org.apache.hadoop.mapreduce.lib.input

 TextInputFormat.


  @Override
  public RecordReader<LongWritable, Text> 
    createRecordReader(InputSplit split,
                       TaskAttemptContext context) {
    return new LineRecordReader();  // 进入
  }
  }


  如下:
  /**
 * Treats keys as offset in file and value as line. 
 */
  LineRecordReader.java
  
  nextKeyValue()  调用一次  则当前 key  value 就被赋值到,  相当于迭代器
  
    public boolean nextKeyValue() throws IOException {
    if (key == null) {
      key = new LongWritable(); // 第一次调用时,key = 0  eg: hello.txt (大小10KB) 
    }
    key.set(pos);
    if (value == null) {
      value = new Text();
    }
    int newSize = 0;
    // We always read one extra line, which lies outside the upper
    // split limit i.e. (end - 1)
    while (getFilePosition() <= end) {
      newSize = in.readLine(value, maxLineLength,  //读取第一行时,将第一行内容写入到value内,并返回第一行读完后的位置
          Math.max(maxBytesToConsume(pos), maxLineLength));
      if (newSize == 0) {
        break;
      }
      pos += newSize; // 记录pos现在的位置,为读取第二行时, key.set(pos);做准备工作
      if (newSize < maxLineLength) {
        break;
      }
	  // 这样赋值对全局变量 key  value赋值完毕后,通过getCurrentKey  getCurrentValue 获取现在的值  


....}





看map如何调用写法:
  
  Mapper.java:
  
   public void run(Context context) throws IOException, InterruptedException {
    setup(context);
    while (context.nextKeyValue()) {// 不断context.nextKeyValue()进入方法后实际就是RecordReader.nextKeyValue()来不断将目标文件的行,行内容放在<k,v>
      map(context.getCurrentKey(), context.getCurrentValue(), context); //不断取值
    }
    cleanup(context);
  }

总结如下:



 

 map函数处理的<k1,v1>从HDFS文件中获取流程源码解析

系统回答如下:

问:从源代码的角度分析map函数处理的<k1,v1>是如何从HDFS文件中获取的?
答:
1.从TextInputFormat入手分析,找到父类FileInputFormat,找到父类InputFormat。
  在InputFormat中找到2个方法,分别是getSplits(...)和createRecordReader(...)。
  通过注释知道getSplits(...)作用是把输入文件集合中的所有内容解析成一个个的InputSplits,每一个InputSplit对应一个mapper task。
  createRecordReader(...)作用是创建一个RecordReader的实现类。RecordReader作用是解析InputSplit产生一个个的<k,v>。
2.在FileInputFormat中找到getSplits(...)的实现。
  通过实现,获知
  (1)每个SplitSize的大小和默认的block大小一致,好处是满足数据本地性。
  (2)每个输入文件都会产生一个InputSplit,即使是空白文件,也会产生InputSPlit;
  如果一个文件非常大,那么会按照InputSplit大小,切分产生多个InputSplit。
3.在TextInputFormat中找到createRecordReader(...)的实现,在方法中找到了LineRecordReader。
  接下来分析LineRecordReader类。  
  在RecordReader类中,通过查看多个方法,知晓key、value作为类的属性存在的,且知道了nextKeyValue()方法的用法。  
  在LineRecordReader类中,重点分析了nextKeyValue(...)方法。在这个方法中,重点分析了newSize = in.readLine(value, maxLineLength, maxBytesToConsume(pos));
  在in.readLine(...)中,第一个形参存储被读取的行文本内容,返回值表示被读取内容的字节数。
  通过以上代码,分析了InputSplit中的内容是如何转化为一个个的<k,v>。
4.从Mapper类中进行分析,发现了setup()、cleanup()、map()、run()。
  在run()方法中,通过while,调用context.nextKeyValue(...)。
  进一步分析Context的接口类是org.apache.hadoop.mapreduce.lib.map.WrappedMapper.MapContext,MapContext调用了nextKeyValue(...)。最终找到了MapContext的实现了MapContextImpl类org.apache.hadoop.mapreduce.task.MapContextImpl。
  在这个类的构造方法中,发现传入了RecordReader的实现类。

相关推荐

硅步至千里 / 0评论 2020-04-19