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虚类的所有方法
相关推荐
自定义inputFormat&&outputFormat1
Hadoop 自定义 Partitioner 实现
Hadoop 自定义 Partitioner 实现
Hadoop 自定义 Partitioner 源代码
hadoop自定义类型编程
hadoop源码,官方,放心下载,完整版,阅读hadoop源代码的目的不一定非是工作的需要,你可以把他看成一种修养,通过阅读hadoop源代码,加深自己对分布式系统的理解,培养自己踏实做事的心态。
基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书...
Hadoop-2.7.2源码及编译jar包.zip,包含 Apache Hadoop 源码包、编译后的 Win7、Win10、以及 Linux 包。 Hadoop-2.7.2源码及编译jar包.zip,包含 Apache Hadoop 源码包、编译后的 Win7、Win10、以及 Linux 包。 ...
实战hadoop,源码,刘鹏,开启通向云计算的捷径
hadoop2.8.4源码
hadoop2.7.3的源码包,hadoop关联源码的时候直接选择就可以查看源码。是自己通过mvn下载的源码之后压缩的。
hadoop 1.2.1核心源码 包括 io,fs,ipc,net,http等类
Hadoop相关资源下载,包括 hadoop-2.7.3源码和Hadoop的2.7.3的安装包,用于Hadoop源码编译和学习。
win 7 64上编译 Hadoop 2.7.3 源码 的真实经历。
Hadoop及源码 Hadoop及源码 Hadoop及源码 Hadoop及源码
Hadoop豆瓣电影数据分析(Hadoop)操作源码
内含Hadoop权威指南第三版和第四版,Hadoop技术内幕Yarn与common与Hdfs架构设计原理,Mapreduce详解等6本书
org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)Z Hadoop源码_NativeIO.java