当前位置:首页 > 开发 > 系统架构 > 架构 > 正文

hadoop自定义inputformat源码

发表于: 2013-02-17   作者:blackproof   来源:转载   浏览次数:
摘要: hadoop的inputformat包括他的子类reader是maptask读取数据的重要步骤 一、获得splits-mapper数 1. jobclinet的submitJobInternal,生成split,获取mapper数量   public RunningJob submitJobInternal { return ugi.doAs(new Privi

hadoop的inputformat包括他的子类reader是maptask读取数据的重要步骤

一、获得splits-mapper数

1. jobclinet的submitJobInternal,生成split,获取mapper数量

 

public 
  RunningJob submitJobInternal {
    return ugi.doAs(new PrivilegedExceptionAction<RunningJob>() {
....
int maps = writeSplits(context, submitJobDir);//生成split,获取mapper数量
....
}}
 jobclinet的writesplit方法

 

 

private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,
      Path jobSubmitDir) throws IOException,
      InterruptedException, ClassNotFoundException {
    JobConf jConf = (JobConf)job.getConfiguration();
    int maps;
    if (jConf.getUseNewMapper()) {
      maps = writeNewSplits(job, jobSubmitDir);//新api调用此方法
    } else {
      maps = writeOldSplits(jConf, jobSubmitDir);
    }
    return maps;
  }
  2.writeNewSplits新api方法,反射inputformat类,调用getsplit方法,获取split数据,并排序,并返回mapper数量
private <T extends InputSplit>
  int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,
      InterruptedException, ClassNotFoundException {
    Configuration conf = job.getConfiguration();
    InputFormat<?, ?> input =
      ReflectionUtils.newInstance(job.getInputFormatClass(), conf);//反射到inputsplit

    List<InputSplit> splits = input.getSplits(job);//调用inputformat子类实现的getsplits方法
    T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);//生成数组,这么简单的方法写的这么复杂,真够扯的,不懂这样为了什么

    // sort the splits into order based on size, so that the biggest
    // go first
    Arrays.sort(array, new SplitComparator());//splits排序
    JobSplitWriter.createSplitFiles(jobSubmitDir, conf,
        jobSubmitDir.getFileSystem(conf), array);
    return array.length;//mapper数量
  }

 

3.贴上最常用的FileInputSplit的getSplits方法

 

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>();
    List<FileStatus>files = listStatus(job);
    for (FileStatus file: files) {
      Path path = file.getPath();
      FileSystem fs = path.getFileSystem(job.getConfiguration());
      long length = file.getLen();
      BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
      if ((length != 0) && isSplitable(job, path)) { 
        long blockSize = file.getBlockSize();
        long splitSize = computeSplitSize(blockSize, minSize, maxSize);//获得split文件的最大文件大小

        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 { 
        //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;
  }
 

 

 

二、读取keyvalue的过程

1.实例化inputformat,初始化reader

在MapTask类的runNewMapper方法中,生成inputformat和recordreader,并进行初始化,运行mapper

MapTask$NewTrackingRecordReader 由 RecordReader组成,是它的一个代理类

 private <INKEY,INVALUE,OUTKEY,OUTVALUE>
  void runNewMapper {
 // 生成自定义inputformat
    org.apache.hadoop.mapreduce.InputFormat<INKEY,INVALUE> inputFormat =
      (org.apache.hadoop.mapreduce.InputFormat<INKEY,INVALUE>)
        ReflectionUtils.newInstance(taskContext.getInputFormatClass(), job);
.....
//生成自定义recordreader
org.apache.hadoop.mapreduce.RecordReader<INKEY,INVALUE> input =
      new NewTrackingRecordReader<INKEY,INVALUE>
          (split, inputFormat, reporter, job, taskContext);
.....
//初始化recordreader
input.initialize(split, mapperContext);
.....
//运行mapper
mapper.run(mapperContext);
   }

 

2.在运行mapper中,调用context让reader读取key和value,其中使用代理类MapTask$NewTrackingRecordReader,添加并推送读取记录

    mapper代码:

 public void run(Context context) throws IOException, InterruptedException {
    setup(context);
     
    while (context.nextKeyValue()) {
      map(context.getCurrentKey(), context.getCurrentValue(), context);
    }
    cleanup(context);
  }

 MapContext代码:

@Override
  public boolean nextKeyValue() throws IOException, InterruptedException {
    return reader.nextKeyValue();
  }
@Override
  public KEYIN getCurrentKey() throws IOException, InterruptedException {
    return reader.getCurrentKey();
  }

  @Override
  public VALUEIN getCurrentValue() throws IOException, InterruptedException {
    return reader.getCurrentValue();
  }

 MapTask$NewTrackingRecordReader的代码:

 @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
      boolean result = false;
      try {
        long bytesInPrev = getInputBytes(fsStats);
        result = real.nextKeyValue();//recordreader实际读取数据
        long bytesInCurr = getInputBytes(fsStats);

        if (result) {
          inputRecordCounter.increment(1);//添加读取记录
          fileInputByteCounter.increment(bytesInCurr - bytesInPrev);//记录读取数据
        }
        reporter.setProgress(getProgress());//将reporter的flag置为true,推送记录信息
      } catch (IOException ioe) {
        if (inputSplit instanceof FileSplit) {
          FileSplit fileSplit = (FileSplit) inputSplit;
          LOG.error("IO error in map input file "
              + fileSplit.getPath().toString());
          throw new IOException("IO error in map input file "
              + fileSplit.getPath().toString(), ioe);
        }
        throw ioe;
      }
      return result;
    }

 3.执行完mapper方法,返回到maptask,关闭reader

  

      mapper.run(mapperContext);
      input.close();//关闭inputformat
      output.close(mapperContext);

 

 两个步骤不在同一个线程中完成,生成splits后进入monitor阶段

以上也调用了所有的inputformat虚类的所有方法

 

 

hadoop自定义inputformat源码

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
在上一篇中,我们实现了按 cookieId 和 time 进行二次排序,现在又有新问题:假如我需要按 cookieId
平时我们写MapReduce程序的时候,在设置输入格式的时候,总会调用形如job.setInputFormatClass(KeyV
<img src="http://img.it610.com/image/product/d20a8897e10949cbb4c1f82d32420b17.png" width="
开放环境,hadoop-0.20.2,hive-0.6 1.日志分隔符 2010-05-31 10:50:17|||61.132.4.82|||http://www.
InputFormat 主要用于描述输入数据的格式, 它提供以下两个功能。 ❑数据切分:按照某个策略将输入
Hadoop MapReduce的编程接口层主要有5个可编程组件,分别为InputFormat、Mapper、Partitioner、Reduc
在执行一个Job的时候,Hadoop会将输入数据划分成N个Split,然后启动相应的N个Map程序来分别处理它们
在执行一个Job的时候,Hadoop会将输入数据划分成N个Split,然后启动相应的N个Map程序来分别处理它们
在执行一个Job的时候,Hadoop会将输入数据划分成N个Split,然后启动相应的N个Map程序来分别处理它们
在本节里,我们着重学习MapReduce编程接口模型中的InputForamt组件。 InputFormat主要用于描述输入
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号