ReduceTask的执行

Reduce处理程序中须要运行三个类型的处理，

1.copy,从各map中copy数据过来

2.sort,对数据进行排序操作。

3.reduce,运行业务逻辑的处理。

ReduceTask的执行也是通过run方法開始，

通过mapreduce.job.reduce.shuffle.consumer.plugin.class配置shuffle的plugin,

默认是Shuffle实现类。实现ShuffleConsumerPlugin接口。

生成Shuffle实例，并运行plugin的init函数进行初始化。

Class<?extendsShuffleConsumerPlugin>clazz =

job.getClass(MRConfig.SHUFFLE_CONSUMER_PLUGIN,Shuffle.class,ShuffleConsumerPlugin.class);

shuffleConsumerPlugin =ReflectionUtils.newInstance(clazz, job);

LOG.info("UsingShuffleConsumerPlugin: " +shuffleConsumerPlugin);

ShuffleConsumerPlugin.ContextshuffleContext =

newShuffleConsumerPlugin.Context(getTaskID(),job, FileSystem.getLocal(job),umbilical,

super.lDirAlloc,reporter, codec,

combinerClass,combineCollector,

spilledRecordsCounter,reduceCombineInputCounter,

shuffledMapsCounter,

reduceShuffleBytes,failedShuffleCounter,

mergedMapOutputsCounter,

taskStatus,copyPhase,sortPhase,this,

mapOutputFile,localMapFiles);

shuffleConsumerPlugin.init(shuffleContext);

运行shuffle的run函数。得到RawKeyValueIterator的实例。

rIter =shuffleConsumerPlugin.run();

Shuffle.run函数定义：

.....................................

inteventsPerReducer = Math.max(MIN_EVENTS_TO_FETCH,

MAX_RPC_OUTSTANDING_EVENTS/ jobConf.getNumReduceTasks());

intmaxEventsToFetch = Math.min(MAX_EVENTS_TO_FETCH,eventsPerReducer);

生成map的完毕状态获取线程，并启动此线程，此线程中从am中获取此job中全部完毕的map的event

通过ShuffleSchedulerImpl实例把全部的map的完毕的map的host,mapid,

等记录到mapLocations容器中。此线程每一秒运行一个获取操作。

//Start the map-completion events fetcher thread

finalEventFetcher<K,V> eventFetcher =

newEventFetcher<K,V>(reduceId,umbilical,scheduler,this,

maxEventsToFetch);

eventFetcher.start();

以下看看EventFetcher.run函数的运行过程：以下代码中我仅仅保留了代码的主体部分。

...................

EventFetcher.run:

publicvoid run() {

intfailures = 0;

........................

intnumNewMaps = getMapCompletionEvents();

..................................

}

......................

}

EventFetcher.getMapCompletionEvents

..................................

MapTaskCompletionEventsUpdateupdate =

umbilical.getMapCompletionEvents(

(org.apache.hadoop.mapred.JobID)reduce.getJobID(),

fromEventIdx,

maxEventsToFetch,

(org.apache.hadoop.mapred.TaskAttemptID)reduce);

events =update.getMapTaskCompletionEvents();

.....................

for(TaskCompletionEvent event : events) {

scheduler.resolve(event);

if(TaskCompletionEvent.Status.SUCCEEDED== event.getTaskStatus()) {

++numNewMaps;

}

}

shecduler是ShuffleShedulerImpl的实例。

ShuffleShedulerImpl.resolve

caseSUCCEEDED:

URI u = getBaseURI(reduceId,event.getTaskTrackerHttp());

addKnownMapOutput(u.getHost() +":"+ u.getPort(),

u.toString(),

event.getTaskAttemptId());

maxMapRuntime= Math.max(maxMapRuntime,event.getTaskRunTime());

break;

.......

ShuffleShedulerImpl.addKnownMapOutput函数：

把mapid与相应的host加入到mapLocations容器中，

MapHost host =mapLocations.get(hostName);

if(host == null){

host = newMapHost(hostName, hostUrl);

mapLocations.put(hostName,host);

}

此时会把host的状设置为PENDING

host.addKnownMap(mapId);

同一时候把host加入到pendingHosts容器中。notify相关的Fetcher文件copy线程。

//Mark the host as pending

if(host.getState() == State.PENDING){

pendingHosts.add(host);

notifyAll();

}

.....................

回到ReduceTask.run函数中，接着向下运行

//Start the map-output fetcher threads

booleanisLocal = localMapFiles!= null;

通过mapreduce.reduce.shuffle.parallelcopies配置的值。默觉得5。生成获取map数据的线程数。

生成Fetcher线程实例，并启动相关的线程。

通过mapreduce.reduce.shuffle.connect.timeout配置连接超时时间。默认180000

通过mapreduce.reduce.shuffle.read.timeout配置读取超时时间。默觉得180000

finalintnumFetchers = isLocal ? 1 :

jobConf.getInt(MRJobConfig.SHUFFLE_PARALLEL_COPIES,5);

Fetcher<K,V>[] fetchers =newFetcher[numFetchers];

if(isLocal) {

fetchers[0] = newLocalFetcher<K, V>(jobConf,reduceId,scheduler,

merger,reporter,metrics,this,reduceTask.getShuffleSecret(),

localMapFiles);

fetchers[0].start();

}else{

for(inti=0; i < numFetchers; ++i) {

fetchers[i] = newFetcher<K,V>(jobConf,reduceId,scheduler,merger,

reporter,metrics,this,

reduceTask.getShuffleSecret());

fetchers[i].start();

}

}

.........................

接下来进行Fetcher线程里面，看看Fetcher.run函数执行流程：

..........................

MapHost host = null;

try{

//If merge is on, block

merger.waitForResource();

从ShuffleScheduler中取出一个MapHost实例，

//Get a host to shuffle from

host = scheduler.getHost();

metrics.threadBusy();

运行shuffle操作。

//Shuffle

copyFromHost(host);

} finally{

if(host != null){

scheduler.freeHost(host);

metrics.threadFree();

}

}

接下来看看ShuffleScheduler中的getHost函数：

........

假设pendingHosts的值没有。先wait住，等待EventFetcher线程去获取数据来notify此wait

while(pendingHosts.isEmpty()){

wait();

}

MapHost host = null;

Iterator<MapHost> iter =pendingHosts.iterator();

从pendingHosts中random出一个MapHost，并返回给调用程序。

intnumToPick = random.nextInt(pendingHosts.size());

for(inti=0; i <= numToPick; ++i) {

host = iter.next();

}

pendingHosts.remove(host);

........................

当得到一个MapHost后，运行copyFromHost来进行数据的copy操作。

此时。一个task的host的url样子基本上是这个样子：

host:port/mapOutput?

job=xxx&reduce=123(当前reduce的partid值)&map=

copyFromHost的代码部分：

.....

List<TaskAttemptID>maps = scheduler.getMapsForHost(host);

.....

Set<TaskAttemptID>remaining = newHashSet<TaskAttemptID>(maps);

.....

此部分完毕后，url样子中map=后面会有非常多个mapid。多个用英文的”,”号分开的。

URLurl = getMapOutputURL(host, maps);

此处依据url打开httpconnection,

假设mapreduce.shuffle.ssl.enabled配置为true时，会打开SSL连接。默觉得false.

openConnection(url);

.....

设置连接超时时间，header,读取超时时间等值。并打开HttpConnection的连接。

// put url hashinto http header

connection.addRequestProperty(

SecureShuffleUtils.HTTP_HEADER_URL_HASH,encHash);

//set the read timeout

connection.setReadTimeout(readTimeout);

//put shuffle version into httpheader

connection.addRequestProperty(ShuffleHeader.HTTP_HEADER_NAME,

ShuffleHeader.DEFAULT_HTTP_HEADER_NAME);

connection.addRequestProperty(ShuffleHeader.HTTP_HEADER_VERSION,

ShuffleHeader.DEFAULT_HTTP_HEADER_VERSION);

connect(connection,connectionTimeout);

.....

运行文件的copy操作。此处是迭代运行。每个读取一个map的文件。

并把remaining中的值去掉一个。直到remaining的值所有读取完毕。

TaskAttemptID[] failedTasks =null;

while(!remaining.isEmpty() && failedTasks == null){

在copyMapOutput函数中。每次读取一个mapid,

依据MergeManagerImpl中的reserve函数，

1.检查map的输出是否超过了mapreduce.reduce.memory.totalbytes配置的大小。

此配置的默认值

是当前Runtime的maxMemory*mapreduce.reduce.shuffle.input.buffer.percent配置的值。

Buffer.percent的默认值为0.90;

假设mapoutput超过了此配置的大小时,生成一个OnDiskMapOutput实例。

2.假设没有超过此大小，生成一个InMemoryMapOutput实例。

failedTasks =copyMapOutput(host, input, remaining);

}

在copyMapOutput函数中首先调用的MergeManagerImpl.reserve函数：

if(!canShuffleToMemory(requestedSize)) {

.....

returnnewOnDiskMapOutput<K,V>(mapId, reduceId,this,requestedSize,

jobConf,mapOutputFile,fetcher, true);

}

.....

if(usedMemory> memoryLimit){

.....,当前使用的memory已经超过了配置的内存使用大小。此时返回null，

把host又一次加入到shuffleScheduler的pendingHosts队列中。

returnnull;

}

returnunconditionalReserve(mapId, requestedSize, true);

生成一个InMemoryMapOutput,并把usedMemory加上此mapoutput的大小。

privatesynchronizedInMemoryMapOutput<K, V> unconditionalReserve(

TaskAttemptID mapId, longrequestedSize, booleanprimaryMapOutput) {

usedMemory+= requestedSize;

returnnewInMemoryMapOutput<K,V>(jobConf,mapId, this,(int)requestedSize,

codec,primaryMapOutput);

}

以下是当usedMemory使用超过了指定的大小后，的处理部分。又一次把host加入到队列中。

例如以下所看到的：copyMapOutput函数

if(mapOutput == null){

LOG.info("fetcher#"+ id + "- MergeManager returned status WAIT ...");

//Notan error but wait to process data.

returnEMPTY_ATTEMPT_ID_ARRAY;

}

此时host中还有没处理完毕的mapoutput,在Fetcher.run中，又一次加入到队列中把此host

if(host != null){

scheduler.freeHost(host);

metrics.threadFree();

}

.........

接下来还是在copyMapOutput函数中，

通过mapoutput也就是merge.reserve函数返回的实例的shuffle函数。

假设mapoutput是InMemoryMapOutput,在调用shuffle时，直接把map输出写入到内存。

假设是OnDiskMapOutput,在调用shuffle时，直接把map的输出写入到local暂时文件里。

....

最后，运行ShuffleScheduler.copySucceeded完毕文件的copy,调用mapout.commit函数。

scheduler.copySucceeded(mapId,host, compressedLength,

endTime -startTime, mapOutput);

并从remaining中移出处理过的mapid,

接下来看看MapOutput.commit函数：

a.InMemoryMapOutput.commit函数：

publicvoidcommit() throwsIOException {

merger.closeInMemoryFile(this);

}

调用MergeManagerImpl.closeInMemoryFile函数:

publicsynchronizedvoidcloseInMemoryFile(InMemoryMapOutput<K,V> mapOutput) {

把此mapOutput实例加入到inMemoryMapOutputs列表中。

inMemoryMapOutputs.add(mapOutput);

LOG.info("closeInMemoryFile-> map-output of size: " +mapOutput.getSize()

+ ",inMemoryMapOutputs.size() -> " +inMemoryMapOutputs.size()

+ ",commitMemory -> " + commitMemory+ ", usedMemory ->"+ usedMemory);

把commitMemory的大小添加当前传入的mapoutput的size大小。

commitMemory+=mapOutput.getSize();

检查是否达到merge的值，

此值是mapreduce.reduce.memory.totalbytes配置

*mapreduce.reduce.shuffle.merge.percent配置的值。

默认是当前Runtime的memory*0.90*0.90

也就是说，仅仅有有新的mapoutput增加，这个检查条件就肯定会达到

//Can hang if mergeThreshold is really low.

if(commitMemory>= mergeThreshold){

.......

把正在进行merge的mapoutput列表加入到一起发起merge操作。

inMemoryMapOutputs.addAll(inMemoryMergedMapOutputs);

inMemoryMergedMapOutputs.clear();

inMemoryMerger.startMerge(inMemoryMapOutputs);

commitMemory= 0L; // Reset commitMemory.

}

假设mapreduce.reduce.merge.memtomem.enabled配置为true,默觉得false

同一时候inMemoryMapOutputs中的mapoutput个数

达到了mapreduce.reduce.merge.memtomem.threshold配置的值，

默认值是mapreduce.task.io.sort.factor配置的值，默觉得100

发起memTomem的merger操作。

if(memToMemMerger!= null){

if(inMemoryMapOutputs.size()>= memToMemMergeOutputsThreshold){

memToMemMerger.startMerge(inMemoryMapOutputs);

}

}

}

MergemanagerImpl.InMemoryMerger.merger函数操作：

在运行inMemoryMerger.startMerge(inMemoryMapOutputs);操作后，会notify此线程，

同一时候运行merger函数：

publicvoidmerge(List<InMemoryMapOutput<K,V>> inputs) throwsIOException {

if(inputs == null|| inputs.size() == 0) {

return;

}

....................

TaskAttemptID mapId =inputs.get(0).getMapId();

TaskID mapTaskId =mapId.getTaskID();

List<Segment<K, V>>inMemorySegments = newArrayList<Segment<K, V>>();

生成InMemoryReader实例，并把传入的容器清空，把生成好后的segment放到到inmemorysegments中。

longmergeOutputSize =

createInMemorySegments(inputs,inMemorySegments,0);

intnoInMemorySegments = inMemorySegments.size();

生成一个输出的文件路径。

Path outputPath =

mapOutputFile.getInputFileForWrite(mapTaskId,

mergeOutputSize).suffix(

Task.MERGED_OUTPUT_PREFIX);

针对输出的暂时文件生成一个Write实例。

Writer<K,V> writer =

newWriter<K,V>(jobConf,rfs,outputPath,

(Class<K>)jobConf.getMapOutputKeyClass(),

(Class<V>)jobConf.getMapOutputValueClass(),

codec,null);

RawKeyValueIterator rIter = null;

CompressAwarePathcompressAwarePath;

try{

LOG.info("Initiatingin-memory merge with " +noInMemorySegments +

"segments...");

此部分与map端的输出没什么差别，得到几个segment的文件的一个iterator,

此部分是一个优先堆，每一次next都会从全部的segment中读取出最小的一个key与value

rIter = Merger.merge(jobConf,rfs,

(Class<K>)jobConf.getMapOutputKeyClass(),

(Class<V>)jobConf.getMapOutputValueClass(),

inMemorySegments,inMemorySegments.size(),

newPath(reduceId.toString()),

(RawComparator<K>)jobConf.getOutputKeyComparator(),

reporter,spilledRecordsCounter,null,null);

假设没有combiner程序，直接写入到文件，否则，假设有combiner，先运行combiner处理。

if(null== combinerClass){

Merger.writeFile(rIter,writer, reporter,jobConf);

} else{

combineCollector.setWriter(writer);

combineAndSpill(rIter,reduceCombineInputCounter);

}

writer.close();

此处与map端的输出不同的地方在这里，这里不写入spillindex文件，

而是生成一个CompressAwarePath，把输出路径,大写和小写入到此实例中。

compressAwarePath = newCompressAwarePath(outputPath,

writer.getRawLength(),writer.getCompressedLength());

LOG.info(reduceId+

"Merge of the " + noInMemorySegments+

"files in-memory complete." +

"Local file is " + outputPath + "of size " +

localFS.getFileStatus(outputPath).getLen());

} catch(IOException e) {

//makesure that we delete the ondiskfile that we created

//earlierwhen we invoked cloneFileAttributes

localFS.delete(outputPath,true);

throwe;

}

此处，把生成的文件加入到onDiskMapOutputs属性中。

并检查此容器中的文件是否达到了mapreduce.task.io.sort.factor配置的值，

假设是，发起disk的merger操作。

//Note the output of the merge

closeOnDiskFile(compressAwarePath);

}

}

上面最后一行的所有定义在以下这里。

publicsynchronizedvoidcloseOnDiskFile(CompressAwarePath file) {

onDiskMapOutputs.add(file);

if(onDiskMapOutputs.size()>= (2 * ioSortFactor- 1)) {

onDiskMerger.startMerge(onDiskMapOutputs);

}

}

b.OnDiskMapOutput.commit函数：

把tmp文件rename到指定的文件夹下，生成一个CompressAwarePath实例。调用上面提到的处理程序。

publicvoidcommit() throwsIOException {

fs.rename(tmpOutputPath,outputPath);

CompressAwarePathcompressAwarePath = newCompressAwarePath(outputPath,

getSize(), this.compressedSize);

merger.closeOnDiskFile(compressAwarePath);

}

MergeManagerImpl.OnDiskMerger.merger函数：

这个函数到如今基本上没有什么能够讲解的东西，注意一点就是，

每merge一个文件后，会把这个merge后的文件路径又一次加入到onDiskMapOutputs容器中。

publicvoidmerge(List<CompressAwarePath> inputs) throwsIOException {

//sanity check

if(inputs == null|| inputs.isEmpty()) {

LOG.info("Noondisk files to merge...");

return;

}

longapproxOutputSize = 0;

intbytesPerSum =

jobConf.getInt("io.bytes.per.checksum",512);

LOG.info("OnDiskMerger:We have " + inputs.size() +

"map outputs on disk. Triggering merge...");

//1. Prepare the list of files to be merged.

for(CompressAwarePath file : inputs) {

approxOutputSize +=localFS.getFileStatus(file).getLen();

}

//add the checksum length

approxOutputSize +=

ChecksumFileSystem.getChecksumLength(approxOutputSize,bytesPerSum);

//2. Start the on-disk merge process

Path outputPath =

localDirAllocator.getLocalPathForWrite(inputs.get(0).toString(),

approxOutputSize,jobConf).suffix(Task.MERGED_OUTPUT_PREFIX);

Writer<K,V> writer =

newWriter<K,V>(jobConf,rfs,outputPath,

(Class<K>)jobConf.getMapOutputKeyClass(),

(Class<V>)jobConf.getMapOutputValueClass(),

codec,null);

RawKeyValueIterator iter = null;

CompressAwarePathcompressAwarePath;

Path tmpDir = newPath(reduceId.toString());

try{

iter = Merger.merge(jobConf,rfs,

(Class<K>)jobConf.getMapOutputKeyClass(),

(Class<V>)jobConf.getMapOutputValueClass(),

codec,inputs.toArray(newPath[inputs.size()]),

true,ioSortFactor,tmpDir,

(RawComparator<K>)jobConf.getOutputKeyComparator(),

reporter,spilledRecordsCounter,null,

mergedMapOutputsCounter,null);

Merger.writeFile(iter,writer, reporter,jobConf);

writer.close();

compressAwarePath = newCompressAwarePath(outputPath,

writer.getRawLength(),writer.getCompressedLength());

} catch(IOException e) {

localFS.delete(outputPath,true);

throwe;

}

closeOnDiskFile(compressAwarePath);

LOG.info(reduceId+

"Finished merging " + inputs.size()+

"map output files on disk of total-size "+

approxOutputSize + "."+

"Local output file is " + outputPath+ " of size "+

localFS.getFileStatus(outputPath).getLen());

}

}

ok，如今map的copy部分运行完毕，回到ShuffleConsumerPlugin的run方法中，

也就是Shuffle的run方法中，接着上面的代码向下分析：

此处等待全部的copy操作完毕。

//Wait for shuffle to complete successfully

while(!scheduler.waitUntilDone(PROGRESS_FREQUENCY)){

reporter.progress();

synchronized(this){

if(throwable!= null){

thrownewShuffleError("error in shuffle in "+ throwingThreadName,

throwable);

}

}

}

假设运行到这一行时。说明全部的mapcopy操作已经完毕，

关闭查找map执行状态的线程与执行copy操作的几个线程。

//Stop the event-fetcher thread

eventFetcher.shutDown();

//Stop the map-output fetcher threads

for(Fetcher<K,V> fetcher : fetchers) {

fetcher.shutDown();

}

//stop the scheduler

scheduler.close();

发am发送状态，通知AM。此时要运行排序操作。

copyPhase.complete();// copy is already complete

taskStatus.setPhase(TaskStatus.Phase.SORT);

reduceTask.statusUpdate(umbilical);

运行最后的merge,事实上在合并全部文件与memory中的数据时。也同一时候会进行排序操作。

//Finish the on-going merges...

RawKeyValueIterator kvIter =null;

try{

kvIter = merger.close();

}catch(Throwable e) {

thrownewShuffleError("Error while doingfinal merge " , e);

}

//Sanity check

synchronized(this){

if(throwable!= null){

thrownewShuffleError("error in shuffle in "+ throwingThreadName,

throwable);

}

}

最后返回这个合并后的iterator实例。

returnkvIter;

Merger也就是MergeManagerImpl.close函数：

publicRawKeyValueIterator close() throwsThrowable {

关闭几个merge的线程。在关闭时会等待现有的merge完毕。

//Wait for on-going merges to complete

if(memToMemMerger!= null){

memToMemMerger.close();

}

inMemoryMerger.close();

onDiskMerger.close();

List<InMemoryMapOutput<K,V>> memory =

newArrayList<InMemoryMapOutput<K, V>>(inMemoryMergedMapOutputs);

inMemoryMergedMapOutputs.clear();

memory.addAll(inMemoryMapOutputs);

inMemoryMapOutputs.clear();

List<CompressAwarePath>disk = newArrayList<CompressAwarePath>(onDiskMapOutputs);

onDiskMapOutputs.clear();

运行终于的merge操作。

returnfinalMerge(jobConf,rfs,memory, disk);

}

最后的一个merge操作

privateRawKeyValueIterator finalMerge(JobConf job, FileSystem fs,

List<InMemoryMapOutput<K,V>>inMemoryMapOutputs,

List<CompressAwarePath>onDiskMapOutputs

)throwsIOException {

LOG.info("finalMergecalled with " +

inMemoryMapOutputs.size() +" in-memory map-outputs and "+

onDiskMapOutputs.size() + "on-disk map-outputs");

finalfloatmaxRedPer =

job.getFloat(MRJobConfig.REDUCE_INPUT_BUFFER_PERCENT,0f);

if(maxRedPer > 1.0 || maxRedPer < 0.0) {

thrownewIOException(MRJobConfig.REDUCE_INPUT_BUFFER_PERCENT+

maxRedPer);

}

得到能够cache到内存的大小,比例通过mapreduce.reduce.input.buffer.percent配置，

intmaxInMemReduce = (int)Math.min(

Runtime.getRuntime().maxMemory()* maxRedPer, Integer.MAX_VALUE);

//merge configparams

Class<K> keyClass =(Class<K>)job.getMapOutputKeyClass();

Class<V> valueClass =(Class<V>)job.getMapOutputValueClass();

booleankeepInputs = job.getKeepFailedTaskFiles();

finalPath tmpDir = newPath(reduceId.toString());

finalRawComparator<K> comparator =

(RawComparator<K>)job.getOutputKeyComparator();

//segments required to vacate memory

List<Segment<K,V>>memDiskSegments = newArrayList<Segment<K,V>>();

longinMemToDiskBytes = 0;

booleanmergePhaseFinished = false;

if(inMemoryMapOutputs.size() > 0) {

TaskID mapId =inMemoryMapOutputs.get(0).getMapId().getTaskID();

这个地方依据可cache到内存的值。把不能cache到内存的部分生成InMemoryReader实例。

并加入到memDiskSegments容器中。

inMemToDiskBytes =createInMemorySegments(inMemoryMapOutputs,

memDiskSegments,

maxInMemReduce);

finalintnumMemDiskSegments = memDiskSegments.size();

把内存中多于部分的mapoutput数据写入到文件里，并把文件路径加入到onDiskMapOutputs容器中。

if(numMemDiskSegments > 0 &&

ioSortFactor> onDiskMapOutputs.size()) {

...........

此部分主要是写入内存中多于的mapoutput到磁盘中去

mergePhaseFinished = true;

//must spill to disk, but can‘t retain in-memfor intermediate merge

finalPath outputPath =

mapOutputFile.getInputFileForWrite(mapId,

inMemToDiskBytes).suffix(

Task.MERGED_OUTPUT_PREFIX);

finalRawKeyValueIterator rIter = Merger.merge(job,fs,

keyClass, valueClass,memDiskSegments, numMemDiskSegments,

tmpDir, comparator,reporter,spilledRecordsCounter,null,

mergePhase);

Writer<K,V> writer = newWriter<K,V>(job, fs, outputPath,

keyClass, valueClass, codec,null);

try{

Merger.writeFile(rIter,writer, reporter,job);

writer.close();

onDiskMapOutputs.add(newCompressAwarePath(outputPath,

writer.getRawLength(),writer.getCompressedLength()));

writer = null;

//add to list of final disk outputs.

} catch(IOException e) {

if(null!= outputPath) {

try{

fs.delete(outputPath,true);

} catch(IOException ie) {

//NOTHING

}

}

throwe;

} finally{

if(null!= writer) {

writer.close();

}

}

LOG.info("Merged" + numMemDiskSegments + "segments, " +

inMemToDiskBytes + "bytes to disk to satisfy " +

"reducememory limit");

inMemToDiskBytes = 0;

memDiskSegments.clear();

} elseif(inMemToDiskBytes != 0) {

LOG.info("Keeping" + numMemDiskSegments + "segments, " +

inMemToDiskBytes + "bytes in memory for " +

"intermediate,on-disk merge");

}

}

//segments on disk

List<Segment<K,V>>diskSegments = newArrayList<Segment<K,V>>();

longonDiskBytes = inMemToDiskBytes;

longrawBytes = inMemToDiskBytes;

生成眼下文件里有的全部的mapoutput路径的onDisk数组

CompressAwarePath[] onDisk =onDiskMapOutputs.toArray(

newCompressAwarePath[onDiskMapOutputs.size()]);

for(CompressAwarePath file : onDisk) {

longfileLength = fs.getFileStatus(file).getLen();

onDiskBytes += fileLength;

rawBytes +=(file.getRawDataLength() > 0) ?

file.getRawDataLength() :fileLength;

LOG.debug("Diskfile: " + file + "Length is " + fileLength);

把如今reduce端接收过来并存储到文件里的mapoutput生成segment并加入到distSegments容器中

diskSegments.add(newSegment<K, V>(job, fs, file, codec,keepInputs,

(file.toString().endsWith(

Task.MERGED_OUTPUT_PREFIX)?

null: mergedMapOutputsCounter),file.getRawDataLength()

));

}

LOG.info("Merging" + onDisk.length+ " files, "+

onDiskBytes + "bytes from disk");

按内容的大小从小到大排序此distSegments容器

Collections.sort(diskSegments,newComparator<Segment<K,V>>() {

publicintcompare(Segment<K, V> o1, Segment<K, V> o2) {

if(o1.getLength() == o2.getLength()) {

return0;

}

returno1.getLength() < o2.getLength() ? -1 : 1;

}

});

把如今memory中全部的mapoutput内容生成segment并加入到finalSegments容器中。

//build final list of segments from merged backed by disk + in-mem

List<Segment<K,V>>finalSegments = newArrayList<Segment<K,V>>();

longinMemBytes = createInMemorySegments(inMemoryMapOutputs,

finalSegments,0);

LOG.info("Merging" + finalSegments.size() + "segments, " +

inMemBytes + "bytes from memory into reduce");

if(0 != onDiskBytes) {

finalintnumInMemSegments = memDiskSegments.size();

diskSegments.addAll(0,memDiskSegments);

memDiskSegments.clear();

//Pass mergePhase only if there is a going to be intermediate

//merges. See comment where mergePhaseFinished is being set

Progress thisPhase =(mergePhaseFinished) ? null: mergePhase;

这个部分是把如今磁盘上的mapoutput生成一个iterator,

RawKeyValueIterator diskMerge =Merger.merge(

job, fs, keyClass, valueClass,codec,diskSegments,

ioSortFactor,numInMemSegments, tmpDir, comparator,

reporter,false,spilledRecordsCounter,null,thisPhase);

diskSegments.clear();

if(0 == finalSegments.size()) {

returndiskMerge;

}

把如今磁盘上的iterator也相同加入到finalSegments容器中。

也就是此时，这个容器中有两个优先堆排序的队列，每next一次，要从内存与磁盘中找出最小的一个kv.

finalSegments.add(newSegment<K,V>(

newRawKVIteratorReader(diskMerge, onDiskBytes), true,rawBytes));

}

returnMerger.merge(job,fs, keyClass, valueClass,

finalSegments,finalSegments.size(), tmpDir,

comparator, reporter,spilledRecordsCounter,null,

null);

}

shuffle部分如今所有运行完毕。又一次加到ReduceTask.run函数中，接着代码向下分析：

rIter =shuffleConsumerPlugin.run();

............

RawComparatorcomparator = job.getOutputValueGroupingComparator();

if(useNewApi) {

runNewReducer(job, umbilical,reporter, rIter, comparator,

keyClass,valueClass);

}else{

runOldReducer........

}

在以上代码中运行runNewReducer主要是运行reduce的run函数。

org.apache.hadoop.mapreduce.TaskAttemptContexttaskContext =

neworg.apache.hadoop.mapreduce.task.TaskAttemptContextImpl(job,

getTaskID(), reporter);

//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 =

newNewTrackingRecordWriter<OUTKEY, OUTVALUE>(this,taskContext);

job.setBoolean("mapred.skip.on",isSkipping());

job.setBoolean(JobContext.SKIP_RECORDS,isSkipping());

org.apache.hadoop.mapreduce.Reducer.Context

reducerContext =createReduceContext(reducer, job, getTaskID(),

rIter,reduceInputKeyCounter,

reduceInputValueCounter,

trackedRW,

committer,

reporter,comparator, keyClass,

valueClass);

try{

reducer.run(reducerContext);

}finally{

trackedRW.close(reducerContext);

}

以上代码中创建Reducer执行的Context,并执行reducer.run函数：

createReduceContext函数定义部分代码：

org.apache.hadoop.mapreduce.ReduceContext<INKEY,INVALUE, OUTKEY, OUTVALUE>

reduceContext =

newReduceContextImpl<INKEY, INVALUE, OUTKEY, OUTVALUE>(job,taskId,

rIter,

inputKeyCounter,

inputValueCounter,

output,

committer,

reporter,

comparator,

keyClass,

valueClass);

org.apache.hadoop.mapreduce.Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE>.Context

reducerContext =

newWrappedReducer<INKEY, INVALUE, OUTKEY,OUTVALUE>().getReducerContext(

reduceContext);

ReduceContextImpl主要是运行在RawKeyValueInterator中读取数据的相关操作。

Reducer.run函数：

publicvoid run(Context context) throwsIOException, InterruptedException {

setup(context);

try{

while(context.nextKey()) {

reduce(context.getCurrentKey(),context.getValues(),context);

//If a back up store is used, reset it

Iterator<VALUEIN> iter =context.getValues().iterator();

if(iterinstanceofReduceContext.ValueIterator) {

((ReduceContext.ValueIterator<VALUEIN>)iter).resetBackupStore();

}

}

}finally{

cleanup(context);

}

}

在run函数中通过context.nextkey来得到下一行的数据，这部分主要在ReduceContextImpl中完毕：

nextkey调用nextKeyValue函数：

publicboolean nextKeyValue() throwsIOException, InterruptedException {

if(!hasMore){

key= null;

value= null;

returnfalse;

}

此处用来检查是否是一个key以下的第一个value,假设是第一个value时。此值为false,

也就是说。nextKeyIsSame的值是true时，表示如今next的数据与current的key是一行数据。

否则表示已经进行了换行操作。

firstValue= !nextKeyIsSame;

运行一下RawKeyValueInterator(也就是Merge中的队列)，得到当前最小的key

DataInputBuffer nextKey =input.getKey();

把key设置到buffer中，设置到buffer中的目的是为了通过keyDeserializer来读取一个key的值。

currentRawKey.set(nextKey.getData(),nextKey.getPosition(),

nextKey.getLength()- nextKey.getPosition());

buffer.reset(currentRawKey.getBytes(),0, currentRawKey.getLength());

从buffer中读取key的值，并存储到key中，这个地方要注意一下，

以下先看看这部分的定义：

.........................

生成一个key的Deserializer实例，

this.keyDeserializer= serializationFactory.getDeserializer(keyClass);

把buffer当成keyDeserializer的InputStream。

this.keyDeserializer.open(buffer);

Deserializer中运行deserializer函数的定义：

此部分定义能够看出，一个key/value仅仅会生成实例，此部分从性能上考虑主要是为了降低对象的生成。

每次生成一个数据时，都是通过readFields又一次去生成Writable实例中的内容。

因此，非常多同学在reduce中使用value时，会出现数据引用不正确的情况，

由于对象还是同一个对象，但值是最后一个，所以会出现数据不正确的情况

publicWritable deserialize(Writable w) throwsIOException {

Writable writable;

if(w == null){

writable

= (Writable)ReflectionUtils.newInstance(writableClass,getConf());

} else{

writable = w;

}

writable.readFields(dataIn);

returnwritable;

}

.........................

读取key的内容

key= keyDeserializer.deserialize(key);

按key同样的方式，得到当前的value的值。

DataInputBuffer nextVal =input.getValue();

buffer.reset(nextVal.getData(),nextVal.getPosition(), nextVal.getLength()

- nextVal.getPosition());

value= valueDeserializer.deserialize(value);

currentKeyLength= nextKey.getLength() - nextKey.getPosition();

currentValueLength= nextVal.getLength() - nextVal.getPosition();

isMarked的值为false,同一时候backupStore属性为null

if(isMarked){

backupStore.write(nextKey,nextVal);

}

把input运行一次next操作，此处会从全部的文件/memory中找到最小的一个kv.

hasMore= input.next();

if(hasMore) {

比較一下，是否与currentkey是同一个key,假设是表示在同一行中。

也就是key同样。

nextKey = input.getKey();

nextKeyIsSame= comparator.compare(currentRawKey.getBytes(),0,

currentRawKey.getLength(),

nextKey.getData(),

nextKey.getPosition(),

nextKey.getLength()- nextKey.getPosition()

)== 0;

}else{

nextKeyIsSame= false;

}

inputValueCounter.increment(1);

returntrue;

}

接下来是调用reduce函数。此时会通过context.getValues函数把key相应的全部的value传给reduce.

此处的context.getValues例如以下所看到的：

ReduceContextImpl.getValues()

public

Iterable<VALUEIN>getValues() throwsIOException, InterruptedException {

returniterable;

}

以上代码中直接返回的是iterable的实例，此实例在ReduceContextImpl实例生成时生成。

privateValueIterable iterable = newValueIterable();

这个类是ReduceContextImpl中的内部类

protectedclass ValueIterable implementsIterable<VALUEIN> {

privateValueIterator iterator= newValueIterator();

@Override

publicIterator<VALUEIN> iterator() {

returniterator;

}

}

此实例中引用一个ValueIterator类。这也是一个内部类。

每次进行运行时。通过此ValueIterator.next来获取一条数据，

publicVALUEIN next() {

inReset的值默觉得false.也就是说inReset检查内部的代码不会运行，事实上backupStore本身值就是null

假设想使用backupStore,须要运行其内部的make函数。

if(inReset) {

.................里面的代码不分析

}

假设是key以下的第一个value,把firstValue设置为false,由于下一次来时，就不是firstValue了.

返回当前的value

//if this is the first record, we don‘t need to advance

if(firstValue){

firstValue= false;

returnvalue;

}

//if this isn‘t the first record and the next key is different, they

//can‘t advance it here.

if(!nextKeyIsSame){

thrownewNoSuchElementException("iteratepast last value");

}

//otherwise, go to the next key/value pair

try{

这里表示不是第一个value的时候，也就是firstValue的值为false,运行一下nextKeyValue函数。

得到当前的value.返回。

nextKeyValue();

returnvalue;

} catch(IOException ie) {

thrownewRuntimeException("next valueiterator failed", ie);

} catch(InterruptedException ie) {

//this is bad, but we can‘t modify the exception list of java.util

thrownewRuntimeException("next valueiterator interrupted", ie);

}

}

当reduce运行完毕后的输出。跟map端无reduce时的输出一样。

直接输出。