hadoop outputformat是reduceTask中的重要过程
1.实例化outputformat,检查输出目录合法性
在jobClient的submitJobInternal反射生成的outputformat
// Check the output specification if (reduces == 0 ? jobCopy.getUseNewMapper() : jobCopy.getUseNewReducer()) { org.apache.hadoop.mapreduce.OutputFormat<?,?> output = ReflectionUtils.newInstance(context.getOutputFormatClass(), jobCopy);//生成outputformat output.checkOutputSpecs(context); } else { jobCopy.getOutputFormat().checkOutputSpecs(fs, jobCopy); }
贴上一个最常用的FileOutputFormat的checkOutputSpaces的方法
// Ensure that the output directory is set and not already there Path outDir = getOutputPath(job);//获得mapred.output.dir的目录 if (outDir == null) { throw new InvalidJobConfException("Output directory not set."); } // get delegation token for outDir's file system TokenCache.obtainTokensForNamenodes(job.getCredentials(), new Path[] {outDir}, job.getConfiguration()); if (outDir.getFileSystem(job.getConfiguration()).exists(outDir)) {//获得当前job的fs,判断目录是否存在 throw new FileAlreadyExistsException("Output directory " + outDir + " already exists"); }
写出key和value
1.生成inputformat和recordwritter
Task中的initialize方法,创建outputformat,并生成committer,这样mapper和reducer都会执行
主要在ReducerTask中使用outputformat,在runNewReducer方法中,获取recordWritrer
// make a task context so we can get the classes org.apache.hadoop.mapreduce.TaskAttemptContext taskContext = new org.apache.hadoop.mapreduce.TaskAttemptContext(job, getTaskID()); // make a reducer org.apache.hadoop.mapreduce.Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE> reducer = (org.apache.hadoop.mapreduce.Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE>) ReflectionUtils.newInstance(taskContext.getReducerClass(), job); org.apache.hadoop.mapreduce.RecordWriter<OUTKEY,OUTVALUE> trackedRW = new NewTrackingRecordWriter<OUTKEY, OUTVALUE>(reduceOutputCounter, job, reporter, taskContext);//NewTrackingRecordWriter一样也是recordWriter的代理类 job.setBoolean("mapred.skip.on", isSkipping());
2.写出key和value
在自定义Reducer运行run方法中,调用reducer进行业务处理
public void run(Context context) throws IOException, InterruptedException { setup(context); while (context.nextKey()) { reduce(context.getCurrentKey(), context.getValues(), context);//执行reduce } cleanup(context); }
在reducer的reduce方法,使用Reducer$Context调用自定义recordWriter的代理类
Reducer$Context代码:
/** * Generate an output key/value pair. */ public void write(KEYOUT key, VALUEOUT value ) throws IOException, InterruptedException { output.write(key, value); }
NewTrackingRecordWriter代码:
@Override public void write(K key, V value) throws IOException, InterruptedException { long bytesOutPrev = getOutputBytes(fsStats); real.write(key,value); long bytesOutCurr = getOutputBytes(fsStats); fileOutputByteCounter.increment(bytesOutCurr - bytesOutPrev); outputRecordCounter.increment(1); }
最终在reducerTask中关闭writter
reducer.run(reducerContext); trackedRW.close(reducerContext);
相关推荐
MapReduce之自定义 OutPutFormat,通过一个案例,实现自定义的一个OutPutFormat,来更加的深刻的理解MR的过程
Hadoop 自定义 Partitioner 实现
Hadoop 自定义 Partitioner 实现
Hadoop 自定义 Partitioner 源代码
hadoop自定义类型编程
hadoop源码,官方,放心下载,完整版,阅读hadoop源代码的目的不一定非是工作的需要,你可以把他看成一种修养,通过阅读hadoop源代码,加深自己对分布式系统的理解,培养自己踏实做事的心态。
基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书推荐系统源码+数据库.zip基于Hadoop图书...
实战hadoop,源码,刘鹏,开启通向云计算的捷径
Hadoop-2.7.2源码及编译jar包.zip,包含 Apache Hadoop 源码包、编译后的 Win7、Win10、以及 Linux 包。 Hadoop-2.7.2源码及编译jar包.zip,包含 Apache Hadoop 源码包、编译后的 Win7、Win10、以及 Linux 包。 ...
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