大数据（Mapper输出缓冲区MapOutputBuffer） _大数据

一年好景君须记，最是橙黄橘绿时。这篇文章主要讲述大数据：Mapper输出缓冲区MapOutputBuffer相关的知识，希望能为你提供帮助。
Mapper的输出缓冲区MapOutputBuffer
现在我们知道了Map的输入端，紧接着我们看map的输出，这里重点就是context.write这个语句的内涵。获取视频中文档资料及完整视频的伙伴请加QQ群：947967114
搞清Mapper作为参数传给map的context，这里我们看Mapper的run被调用的时候作为了参数传递下来。调用Mapper.run的是MapTask. runNewMapper。到这里我们深究一下runNewMapper。我们看MapTask的run方法：我们重点看runNewMapper
public void run(final JobConf job, final TaskUmbilicalProtocol umbilical)

throws IOException, ClassNotFoundException, InterruptedException {this.umbilical = umbilical; if (isMapTask()) {// If there are no reducers then there won‘t be any sort. Hence the map// phase will govern the entire attempt‘s progress.if (conf.getNumReduceTasks() == 0) {mapPhase = getProgress().addPhase("map", 1.0f); } else {// If there are reducers then the entire attempt‘s progress will be// split between the map phase (67%) and the sort phase (33%).mapPhase = getProgress().addPhase("map", 0.667f); sortPhase= getProgress().addPhase("sort", 0.333f); }}TaskReporter reporter = startReporter(umbilical); 获取视频中文档资料及完整视频的伙伴请加QQ群：947967114boolean useNewApi = job.getUseNewMapper(); initialize(job, getJobID(), reporter, useNewApi); // check if it is a cleanupJobTaskif (jobCleanup) {runJobCleanupTask(umbilical, reporter); return; }if (jobSetup) {runJobSetupTask(umbilical, reporter); return; }if (taskCleanup) {runTaskCleanupTask(umbilical, reporter); return; }if (useNewApi) {runNewMapper(job, splitMetaInfo, umbilical, reporter); } else {runOldMapper(job, splitMetaInfo, umbilical, reporter); }done(umbilical, reporter);

}
当我们点runNewMapper的时候就能进入真正实现：
private < INKEY,INVALUE,OUTKEY,OUTVALUE>
void runNewMapper(final JobConf job,final TaskSplitIndex splitIndex,final TaskUmbilicalProtocol umbilical,TaskReporter reporter) throws IOException, ClassNotFoundException,
InterruptedException {

// make a task context so we can get the classesorg.apache.hadoop.mapreduce.TaskAttemptContext taskContext =new org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl(job,getTaskID(),reporter); // make a mapperorg.apache.hadoop.mapreduce.Mapper< INKEY,INVALUE,OUTKEY,OUTVALUE> mapper =(org.apache.hadoop.mapreduce.Mapper< INKEY,INVALUE,OUTKEY,OUTVALUE> )

ReflectionUtils.newInstance(taskContext.getMapperClass(), job);
//确定该用哪一种具体的Mapper，然后创建。获取视频中文档资料及完整视频的伙伴请加QQ群：947967114

org.apache.hadoop.mapreduce.InputFormat< INKEY,INVALUE> inputFormat =(org.apache.hadoop.mapreduce.InputFormat< INKEY,INVALUE> )

ReflectionUtils.newInstance(taskContext.getInputFormatClass(), job);
//确定输入的文件格式

// rebuild the input splitorg.apache.hadoop.mapreduce.InputSplit split = null;

split = getSplitDetails(new Path(splitIndex.getSplitLocation()),splitIndex.getStartOffset()); //确定这个Mapper所用的输入是哪一个split

LOG.info("Processing split: " + split); org.apache.hadoop.mapreduce.RecordReader< INKEY,INVALUE> input =new NewTrackingRecordReader< INKEY,INVALUE> (split, inputFormat, reporter, taskContext);

//创建和InputFormat相称的RecordReader

job.setBoolean(JobContext.SKIP_RECORDS, isSkipping()); org.apache.hadoop.mapreduce.RecordWriter output = null; // get an output object

if (job.getNumReduceTasks() == 0) {
//如果设置的reduce个数是0，就直接输出。

output =new NewDirectOutputCollector(taskContext, job, umbilical, reporter); } else {output = new NewOutputCollector(taskContext, job, umbilical, reporter);

}
接下来我们看一下NewOutputCollector源码获取视频中文档资料及完整视频的伙伴请加QQ群：947967114

NewOutputCollector(org.apache.hadoop.mapreduce.JobContext jobContext,JobConf job,TaskUmbilicalProtocol umbilical,TaskReporter reporter) throws IOException, ClassNotFoundException {collector = createSortingCollector(job, reporter);

//创建通向排序阶段的collecter

partitions = jobContext.getNumReduceTasks();

//通过获取Reduce数量来获得partitions数量。两个数量一一对应

if (partitions > 1) {

//获取的partitions 数量大于1

partitioner = (org.apache.hadoop.mapreduce.Partitioner< K,V> )ReflectionUtils.newInstance(jobContext.getPartitionerClass(), job);

//ReflectionUtils.newInstance创建用户设置的Partitioner，里边的参数jobContext.getPartitionerClass()是对抽象类的某种扩充，表示自己可以书写一个Partitioner类，通过这个方法来获取，如果没有自己写，就是用默认的HashPartitioner

} else {partitioner = new org.apache.hadoop.mapreduce.Partitioner< K,V> () {@Overridepublic int getPartition(K key, V value, int numPartitions) {return partitions - 1; }//只有一个partition就动态扩充抽象类Partitioner类}; }

}
回到runNewMapper源码：

org.apache.hadoop.mapreduce.MapContext< INKEY, INVALUE, OUTKEY, OUTVALUE> mapContext =new MapContextImpl< INKEY, INVALUE, OUTKEY, OUTVALUE> (job, getTaskID(), input, output, committer, reporter, split);

//创建一个用于Mapper的Context。
org.apache.hadoop.mapreduce.Mapper< INKEY,INVALUE,OUTKEY,OUTVALUE> .Context

mapperContext =new WrappedMapper< INKEY, INVALUE, OUTKEY, OUTVALUE> ().getMapContext(mapContext);

//把上边创建的mapContext通过getMapContext获取过来最终传递给mapperContext ，我们继续看getMapContext源码
public Mapper< KEYIN, VALUEIN, KEYOUT, VALUEOUT> .Context
getMapContext(MapContext< KEYIN, VALUEIN, KEYOUT, VALUEOUT> mapContext) {

return new Context(mapContext);

}
//这里返回了Context对象，在查看Context对象。获取视频中文档资料及完整视频的伙伴请加QQ群：947967114

public Context(MapContext< KEYIN, VALUEIN, KEYOUT, VALUEOUT> mapContext) {this.mapContext = mapContext;

}
//我们看到获取了mapContext 的值。所以我们知道WrappedMapper--> Context--> mapContext是一个MapContextImpl。

try {input.initialize(split, mapperContext);

//初始化input，input是recordReader对象，split和mapperContext作为参数
mapper.run(mapperContext);
//我们知道这个run方法运行的是Mapper的run方法，所以看一下这个run
public void run(Context context) throws IOException, InterruptedException {
setup(context);
//获取context

try {while (context.nextKeyValue()) {

//通过nextKeyValue来控制运行

map(context.getCurrentKey(), context.getCurrentValue(), context);

//运行了map方法，给了recordReader提供过来的键值对。

}} finally {cleanup(context); }

}
回到MapTask源码

mapPhase.complete();

//上锁

setPhase(TaskStatus.Phase.SORT);

//所有的task结果进行排序

statusUpdate(umbilical);

//更新runNewMapper状态。

input.close();

//关闭输入流

input = null; output.close(mapperContext);

//关闭输出流

output = null; } finally {closeQuietly(input); closeQuietly(output, mapperContext); }

}
对于输入格式和分片以前已经详细说过了，需要注意NewTrackingRecordReader。我们知道有了InputFormat之后需要创建与他对应的RecordReader。但是在RecordReader上是用NewTrackingRecordReader。不同之处在于Tracking，是一个跟踪，对RecordReader的跟踪，他这里有一个参数reporter，就是用来上报跟踪结果的，RecordReader则没有这个功能。
和输出有关的是collecter，是输出数据的收集器，context.write最后就通过RecodWriter落实到collector.collect上。RecordWriter和RecordReader是同一个层次。RecodWriter是hadoop定义个一个抽象类，具体的RecodWriter就是对这个抽象类的扩充。用于maptask的就是NewDrictDoutputCollecter和NewOutputCollecter。
这两个类叫做OutputCollecter，实际上都是RecordWriter。Collecter只是一种语意的描述。从Mapper的角度看是Writer，是输出。从框架或下游的角度看是Collect，是收集。
如果reducer数量是0，就是没有reducer，Mapper的输出就是整个MR的输出，这个时候用RecordWriter的NewDrictDoutputCollecter，直接输出。相反至少有一个Reducer，那么使用的就是RecordWriter的NewOutputCollecter。这是我们注重的重点内容。我们看NewOutputCollecter源码。定义了几个内容：

collector = createSortingCollector(job, reporter);

//实现MapOutputCollector

partitions = jobContext.getNumReduceTasks();

//负责Mapper输出的分区

partitioner = (org.apache.hadoop.mapreduce.Partitioner< K,V> )

//分发目标的个数，也就是Reducer的个数。

@Overridepublic void write(K key, V value) throws IOException, InterruptedException {collector.collect(key, value,partitioner.getPartition(key, value, partitions));

}
//write只写不读。

@Overridepublic void close(TaskAttemptContext context) throws IOException,InterruptedException {try {collector.flush(); } catch (ClassNotFoundException cnf) {throw new IOException("can‘t find class ", cnf); }collector.close(); }

}
NewOutputCollector分成两部分，一个是collecter还有一个是partitioner。collecter负责实际收集Mapper输出并交付给Reducer的工作，partitioner负责决定把具体的输出交给哪一个Reducer。
有多个Reducer存在，MR框架需要把每个Mapper的每项输出，也就是收集到的所有的KV对。按照某种条件（就是Partioner的实现方式，默认就是HashPartitioner）输出到不同的Reducer。这样就把Mapper的输出划分成了多个分区（Partition），有几个Reducer就把每个Mapper还分成几个Partition，Partitioner就是起到划分的作用。hash的方式。。。。。。。。。。。。
所以在创建NewOutputCollector的构造函数中，就要把具体的collector和partitioner创建好。
hadoop的源码中定义了MapOutputCollector。凡是实现了这个类，除了init和close方法外，还必须提供collect和flush这两个函数，从NewOutputCollector知道这两个函数的调用者是collector，创建collector的方式是通过createSortingCollector来完成的。并且还实现了对KV对的排序。从属关系如下：
YarnChild.main-> PrivilegeExceptionAction.run-> Maptask.run--> RunNewMapper-> NewOutputCollector-> MapTask.createSortingCollector
那么我们来看一下createSortingCollector源码。获取视频中文档资料及完整视频的伙伴请加QQ群：947967114
private < KEY, VALUE> MapOutputCollector< KEY, VALUE>

createSortingCollector(JobConf job, TaskReporter reporter)throws IOException, ClassNotFoundException {MapOutputCollector.Context context =new MapOutputCollector.Context(this, job, reporter); Class< ?> [] collectorClasses = job.getClasses(JobContext.MAP_OUTPUT_COLLECTOR_CLASS_ATTR, MapOutputBuffer.class);

//如果没有添加设置就默认使用MapOutputBuffer.class

int remainingCollectors = collectorClasses.length;

for (Class clazz : collectorClasses) {
//逐一实验设置的collectorClasses

try {if (!MapOutputCollector.class.isAssignableFrom(clazz)) {throw new IOException("Invalid output collector class: " + clazz.getName() +" (does not implement MapOutputCollector)");

//这里告诉我们必须实现MapOutputCollector.class

}Class< ? extends MapOutputCollector> subclazz =clazz.asSubclass(MapOutputCollector.class); LOG.debug("Trying map output collector class: " + subclazz.getName());

//获取日志

MapOutputCollector< KEY, VALUE> collector =ReflectionUtils.newInstance(subclazz, job);

//创建collector对象。

collector.init(context);

//初始化collector，实际上初始化的是MapOutputBuffer对象

LOG.info("Map output collector class = " + collector.getClass().getName()); return collector;

//没有异常就成功了。

} catch (Exception e) {String msg = "Unable to initialize MapOutputCollector " + clazz.getName(); if (--remainingCollectors > 0) {msg += " (" + remainingCollectors + " more collector(s) to try)"; }LOG.warn(msg, e); }}throw new IOException("Unable to initialize any output collector");

}
具体采用什么collector是可以在配置文件mapred-default.xml中设置的，这里的MAP_OUTPUT_COLLECTOR_CLASS_ATTR即mapreduce.job.output.collector.class.如果文件中没有设置就使用默认的MapOutputBuffer。所以实际创建的collcter就是Mapask的MapOutputBuffer。这个类是Maptask的内部类，实现了MapOutputCollector。
可想而知，如果我们另写一个实现了MapOutputCollectior的Collector，并修改配置文件mapred-default.xml中队配置项的设置。那么就可以创建不是MapTask.MapOutputBuffer。那样createSortingCollector创建的就是一个没有排序功能的collector。我们知道MapReduce框架之所以是工作流不是数据流的原因就是因为Mapper和Reducer之间的排序。因为Sort只有在所有数据到来之后才能完成。sort完之后所有数据才被Rducer拉取。那么没有了sort之后代表数据可以不断的流入而不是一次性的填充，MR给我们提供了这种可能性，就是通过写一个不排序的Collector来替代MapOutputBuffer。我们接下来还是把注意力放到runNewMapper上。
当创建了collector和partitioner之后就是Context，MapTask在调用mapper.run时作为参数的是mapperContext，这个对象的类型是WrappedMapper.Context，整个过程是MapContextImpl创建了mapContext对象，通过WrappedMapper对象（是对Mapper的扩充，根据名字就可以知道是对Mapper的包装区别就是在内部定义了Context类），把一个扩充的Mapper.Context包装在Mapper内部，这就是WrappedMapper.Context类对象。下面是部分代码;
public class WrappedMapper< KEYIN, VALUEIN, KEYOUT, VALUEOUT>

extends Mapper< KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

public Mapper< KEYIN, VALUEIN, KEYOUT, VALUEOUT> .Context
getMapContext(MapContext< KEYIN, VALUEIN, KEYOUT, VALUEOUT> mapContext) {

return new Context(mapContext);

}
@InterfaceStability.Evolving
public class Context

extends Mapper< KEYIN, VALUEIN, KEYOUT, VALUEOUT> .Context {protected MapContext< KEYIN, VALUEIN, KEYOUT, VALUEOUT> mapContext;

//MapContext类。被MapContextImpl实现

public Context(MapContext< KEYIN, VALUEIN, KEYOUT, VALUEOUT> mapContext) {this.mapContext = mapContext; }/** * Get the input split for this map. */public InputSplit getInputSplit() {return mapContext.getInputSplit(); }@Overridepublic KEYIN getCurrentKey() throws IOException, InterruptedException {return mapContext.getCurrentKey(); }@Overridepublic VALUEIN getCurrentValue() throws IOException, InterruptedException {return mapContext.getCurrentValue(); }@Overridepublic boolean nextKeyValue() throws IOException, InterruptedException {return mapContext.nextKeyValue(); }

WrappedMapper.Context是对Mapper.Context的扩充。内部mapContext，它的构造函数Context中的this.mapContext就设置成这个MapContextImpl类对象mapContext。WrappedMapper.Context扩充了Mapper.Context的write、getCurrentKey、nextKeyValue等。
传给mapper.run的context就是WrappedMapper.Context对象。里面的mapContext是MapContextImpl对象。
我们继续看Mapper.map的context.write
关系是：MapTask.run-> runNewMapper-> Mapper.run-> Mapper.map
按照这个关系找到了一个没有做任何事的方法。
public void write(KEYOUT key, VALUEOUT value)

throws IOException, InterruptedException;

我们需要找一个实现，这里找到的就是WrappedMapper.Context.write
就是这一段：

public void write(KEYOUT key, VALUEOUT value) throws IOException,InterruptedException {mapContext.write(key, value);

}
这里的调用的其实是MapContextImpl.write。所以我们找到MapContextImpl。当我们看到MapContextImpl源码是看到继承了TaskInputOutputContextImpl我们找到了
public void write(KEYOUT key, VALUEOUT value

) throws IOException, InterruptedException {output.write(key, value);

}
找到这里我们还是没有找到真正的实现，这里的witer实际上调用的是，NewOutputCollector.writer。

public void write(K key, V value) throws IOException, InterruptedException {collector.collect(key, value,partitioner.getPartition(key, value, partitions));

}
绕了一大圈之后我们发现最终回到了NewOutputCollector，这里的write和之前的有明显区别是collect实现的，里面有了分区。我们找的目的是一定要找到write中真正实现了分区写。
我们知道context是个WrappedMappe.Context对象，所以context.write其实就是就是Wrapped.Context.write,这个函数转而调用内部成分mapContext的write函数，而mapContext是个MapContextImpl对象，所以实际调用的是MoapCntextImpl.write。然而MapContextImpl中没有提供write函数，但是我们看到这个类继承了TaskInputOutputContextImpl。所以就继承他的write方法，然后这个write函数调用的是output的write，我们知道这个output参数类型是一个RecordReader，实际上这个output就是MapTask中定义的output，这个output是一个NewOutputCollector，也就是说是调用的NewOutputCollector的write方法，在这个write中我们看到调用了collector的collect，这个collecter就是Maptask.MapOutputBuffer。
【大数据（Mapper输出缓冲区MapOutputBuffer）】在调用Maptask.MapOutputBuffer的collect时增加了一个参数partition，是指明KV去向的，这个值是有job.setPartitionerClass指定的，没有设置就使用了hashPartitioner。下面所有的工作就是由MapTask的MapOutputBuffer来完成了。获取视频中文档资料及完整视频的伙伴请加QQ群：947967114