hadoop2.7作业提交详解之文件分片 hadoop2.7之作业提交详解(上)

编程之家收集整理的这篇文章主要介绍了hadoop2.7作业提交详解之文件分片 hadoop2.7之作业提交详解(上)编程之家小编觉得挺不错的,现在分享给大家,也给大家做个参考。

在前面一篇文章中(hadoop2.7之作业提交详解(上))中涉及到文件的分片。

JobSubmitter.submitJobInternal方法调用
int maps = writeSplits(job,submitJobDir); //设置map的数量,而map的数量是根据文件的大小和分片的大小,以及文件数量决定的

接下来我们看一下JobSubmitter.writeSplits方法

private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,Path jobSubmitDir) throws IOException,InterruptedException,ClassNotFoundException {
  JobConf jConf = (JobConf)job.getConfiguration();
   maps;
  if (jConf.getUseNewMapper()) {
    maps = writeNewSplits(job,jobSubmitDir); //这里我们使用新的方式
  } else {
    maps = writeOldSplits(jConf,jobSubmitDir);
  }
  return maps;
}

接下来继续看JobSubmitter.writeNewSplits方法:

private <T extends InputSplit>
int writeNewSplits(JobContext job,Path jobSubmitDir)  job.getConfiguration();
  InputFormat<?,?> input =
    ReflectionUtils.newInstance(job.getInputFormatClass(),conf);  输入对象,InputFormat是个抽象类  

  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()); 对切片的大小进行排序,最大的放最前面
  JobSplitWriter.createSplitFiles(jobSubmitDir,conf,jobSubmitDir.getFileSystem(conf),array);创建Split文件 
   array.length;
}

接下来看一下InputFormat这个抽象类:

public abstract class InputFormat<K,V> {
    用来返回分片结果
    abstract 
    List<InputSplit> getSplits(JobContext context
                               ) RecordReader是用来从一个输入分片中读取一个一个的K-V对的抽象类,我们可以将其看作是在InputSplit上的迭代器。
    最主要的方法就是nextKeyvalue()方法,由它获取分片上的下一个K-V 对。
     
    RecordReader<K,1)"> createRecordReader(InputSplit split,TaskAttemptContext context
                                        ) 方法都来自于FileInputFormat类,TextInputFormat类只重写了两个方法:如下:

class TextInputFormat extends FileInputFormat<LongWritable,Text> {

  @Override
  public RecordReader<LongWritable,1)"> 
    createRecordReader(InputSplit split,TaskAttemptContext context) {
    String delimiter = context.getConfiguration().get(
        "textinputformat.record.delimiter");
    byte[] recordDelimiterBytes = null;
    if (null != delimiter)
      recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
      LineRecordReader由一个FileSplit构造出来,start是这个FileSplit的起始位置,pos是当前读取分片的位置,
      end是分片结束位置,in是打开的一个读取这个分片的输入流,它是使用这个FileSplit对应的文件名来打开的。
      key和value则分别是每次读取的K-V对。然后我们还看到可以利用getProgress()来跟踪读取分片的进度,
      这个函数就是根据已经读取的K-V对占总K-V对的比例来显示进度的
    return  LineRecordReader(recordDelimiterBytes);
  }

  @Override
  protected boolean isSplitable(JobContext context,Path file) {
 如果是压缩文件就不切分,非压缩文件就切分。
    final CompressionCodec codec =
       CompressionCodecFactory(context.getConfiguration()).getCodec(file);
    null == codec) {
      true;
    }
    return codec instanceof SplittableCompressionCodec;
  }
}

我们在返回到JobSubmitter.writeNewSplits方法中,有List<InputSplit> splits = input.getSplits(job);主要是调用TextInputFormat.getSplits()方法,而TextInputFormat继承了FileInputFormat类,所以调用的就是FileInputFormat.getSplits()方法

public List<InputSplit> getSplits(JobContext job)  IOException {
  StopWatch sw = new StopWatch().start();用来计算纳秒级别的时间
  long minSize = Math.max(getFormatMinSplitSize(),getMinSplitSize(job)); 最小值默认为1
  long maxSize = getMaxSplitSize(job); 最大值为long的最大值,默认为0x7fffffffffffffffL

   generate splits
  List<InputSplit> splits = new ArrayList<InputSplit>();
  List<FileStatus> files = listStatus(job); 获得所有的输入文件
  for (FileStatus file: files) {
    Path path = file.getPath(); 文件路径
    long length = file.getLen(); 文件大小
    if (length != 0) {
      BlockLocation[] blkLocations;
      if (file instanceof LocatedFileStatus) {如果是个含有数据块位置信息的文件 
        blkLocations = ((LocatedFileStatus) file).getBlockLocations();
      } else { 一般文件 
        FileSystem fs = path.getFileSystem(job.getConfiguration());
        blkLocations = fs.getFileBlockLocations(file,0,length);
      }
      if (isSplitable(job,path)) { 判断是否可以分片
        long blockSize = file.getBlockSize(); 128M
        long splitSize = computeSplitSize(blockSize,minSize,maxSize); 计算分片的大小,默认为128M 

        long bytesRemaining = length;
        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) { 判断剩余文件大小是否大于128M*1.1 
          int blkIndex = getBlockIndex(blkLocations,length-bytesRemaining);f返回每个分片起始位置
          splits.add(makeSplit(path,length-bytesRemaining,splitSize,blkLocations[blkIndex].getHosts(),blkLocations[blkIndex].getCachedHosts()));
          bytesRemaining -= splitSize;  依次减去分片的大小,对剩余长度再次分片
        }
 多次分片后,最后的数据长度仍不为0但又不足一个分片大小
        if (bytesRemaining != 0) {
          bytesRemaining);
          splits.add(makeSplit(path,length-不可分,则把整个文件作为一个分片
      }  not splitable
        splits.add(makeSplit(path,length,blkLocations[0].getHosts(),blkLocations[0].getCachedHosts()));
      }
    }  { 
创建空的分片
      Create empty hosts array for zero length files
      splits.add(makeSplit(path,1)">new String[0]));
    }
  }
   Save the number of input files for metrics/loadgen
  job.getConfiguration().setLong(NUM_INPUT_FILES,files.size()); 设置参数NUM_INPUT_FILES
  sw.stop();
   (LOG.isDebugEnabled()) {
    LOG.debug("Total # of splits generated by getSplits: " + splits.size()
        + ",TiMetaken: " + sw.now(TimeUnit.MILLISECONDS));
  }
   splits;
}
public class FileSplit extends InputSplit implements Writable {
  private Path file;输入文件路径 
  private long start;分片在文件中的位置(起点)
  private long length;分片长度
  private String[] hosts;这个分片所在数据块的多个复份所在节点
  private SplitLocationInfo[] hostInfos;每个数据块复份所在节点,以及是否缓存 
}
makeSplit方法存放的分片格式
protected FileSplit makeSplit(Path file,1)">long start,1)">long length,String[] hosts,String[] inMemoryHosts) {
   FileSplit(file,start,hosts,inMemoryHosts);
}

计算分片的大小
long computeSplitSize(long blockSize,1)"> minSize, maxSize) {
   Math.max(minSize,Math.min(maxSize,blockSize));
}

通过FileInputFormat.getSplits(),可以返回一个存放分片的ArraryList,接下继续回到JobSubmitter.writeNewSplits方法中:

接下来将ArrayList转换为数组,并根据分片的大小排序。然后调用JobSplitWriter.createSplitFiles()方法创建split文件。最后返回数组的长度,也就是map的个数。

总结

以上是编程之家为你收集整理的hadoop2.7作业提交详解之文件分片 hadoop2.7之作业提交详解(上)全部内容,希望文章能够帮你解决hadoop2.7作业提交详解之文件分片 hadoop2.7之作业提交详解(上)所遇到的程序开发问题。

如果觉得编程之家网站内容还不错,欢迎将编程之家网站推荐给程序员好友。

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
如您喜欢寻找一群志同道合、互帮互助的学习伙伴,可以点击下方链接加入:
编程之家官方1群:1065694478(已满)
编程之家官方2群:163560250(已满)
编程之家官方3群:312128206

相关文章

猜你在找的大数据相关文章

###windows下链接hadoop集群 1、假如在linux机器上已经搭建好hadoop集群 2、在windows上把hadoop的压缩包解压到一个没有空格的目录下,比如是D盘根目录 3、配置环境
1、zookeeper的安装##配置/etc/hosts192.168.88.130 lgh192.168.88.131 lgh1192.168.88.132 lgh2 ##安装java8 解压配置环
一、迁移背景 服务器出了问题,导致整个cm server界面呈现出不可用的状态,也就是获取不到各个大数据组件以及主机相关的状态的信息,整个cm server的前端界面处于瘫痪的状态,不可用,刚开始怀疑
在前面一篇文章中(hadoop2.7之作业提交详解(上))中涉及到文件的分片。 JobSubmitter.submitJobInternal方法中调用了int maps = writeSplits(j
根据wordcount进行分析: 这上面是个简单wordcount的代码,这里就不一一说明了,我们首先看main方法:获取一个job对象,然后经过一系列的设置,最后调用waitForCompletio
Apache YARN(Yet Another Resource Negotiator)是一个Hadoop集群资源管理系统。YARN是在Hadoop 2引入的,用以改善MapReduce的表现。但是它
数据流 读取文件数据的剖析 为了知道客户端与HDFS,NameNode,DataNode交互过程中数据的流向,请看图3 2,这张图显示了读取文件过程中主要的事件顺序。 客户端通过调用FileSyste
Hive支持自定义map与reduce script。接下来我用一个简单的wordcount例子加以说明。 如果自己使用Java开发,需要处理System.in,System,out以及key/val