首页 > 代码库 > Reduce Task的学习笔记
Reduce Task的学习笔记
MapReduce五大过程已经分析过半了,上次分析完Map的过程,着实花费了我的很多时间,不过收获很大,值得了额,这次用同样的方法分析完了Reduce的过程,也算是彻底摸透了MapReduce思想的2个最最重要的思想了吧。好,废话不多,切入正题,在学习Reduce过程分析的之前,我特意查了书籍上或网络上相关的资料,我发现很大都是大同小异,缺乏对于源码的参照分析,所以我个人认为,我了可以在某些细节上讲得跟明白些,也许会比较好。因为Map和Reduce的过程的整体流程是非常相近的,如果你看过之前我写的Map Task的分析,相信你也能很快理解我的Reduce过程的分析的。Reduce过程的集中表现体现于Reduce Task中,Reduce Task与Map Reduce一样,分为Job-setup Task, Job-cleanup Task, Task-cleanup Task和Reduce Task。我分析的主要是最后一个Reduce Task 。Reduce Task 主要分为5个阶段:
Shuffle------------------->Merge------------------->Sort------------------->Reduce------------------->Write
其中最重要的部分为前3部分,我也会花最多的时间描述前面3个阶段的任务。
Shuffle阶段。我们知道,Reduce的任务在最最开始的时候,就是接收Map任务中输出的中间结果的数据,key-value根据特定的分区算法,给相应的Reduce任务做处理,所以这时需要Reduce任务去远程拷贝Map输出的中间数据了,这个过程就称作Shuffle阶段,所以这个阶段也称为Copy阶段。在Shuffle阶段中,有个GetMapEventsThread,会定期发送RPC请求,获取远程执行好的Map Task的列表,把他们的输出location映射到mapLocation中。
.... //GetMapEventsThread线程是远程调用获得已经完成的Map任务的列表 int numNewMaps = getMapCompletionEvents(); if (LOG.isDebugEnabled()) { if (numNewMaps > 0) { LOG.debug(reduceTask.getTaskID() + ": " + "Got " + numNewMaps + " new map-outputs"); } } Thread.sleep(SLEEP_TIME); }进入getMapCompletionEvents方法,继续看:
... for (TaskCompletionEvent event : events) { switch (event.getTaskStatus()) { case SUCCEEDED: { URI u = URI.create(event.getTaskTrackerHttp()); String host = u.getHost(); TaskAttemptID taskId = event.getTaskAttemptId(); URL mapOutputLocation = new URL(event.getTaskTrackerHttp() + "/mapOutput?job=" + taskId.getJobID() + "&map=" + taskId + "&reduce=" + getPartition()); List<MapOutputLocation> loc = mapLocations.get(host); if (loc == null) { loc = Collections.synchronizedList (new LinkedList<MapOutputLocation>()); mapLocations.put(host, loc); } //loc中添加新的已经完成的,mapOutputLocation,mapLocations是全局共享的 loc.add(new MapOutputLocation(taskId, host, mapOutputLocation)); numNewMaps ++; } break; ....为了避免出现网络热点,Reduce Task对输出的位置进行了混洗的操作,然后保存到scheduleCopies中,后续的拷贝操作都是围绕着这个列表进行的。这个变量保存在了一个叫ReduceCopier的类里面。确认拷贝的目标位置,还只是Shuffle阶段的前半部分,这时看一下,执行的入口代码在哪里。回到Reduce Task的入口run()代码:
public void run(JobConf job, final TaskUmbilicalProtocol umbilical) throws IOException, InterruptedException, ClassNotFoundException { this.umbilical = umbilical; job.setBoolean("mapred.skip.on", isSkipping()); if (isMapOrReduce()) { //设置不同阶段任务的进度 copyPhase = getProgress().addPhase("copy"); sortPhase = getProgress().addPhase("sort"); reducePhase = getProgress().addPhase("reduce"); } // start thread that will handle communication with parent //创建Task任务报告,与父进程进行联系沟通 TaskReporter reporter = new TaskReporter(getProgress(), umbilical, jvmContext); reporter.startCommunicationThread(); //判断是否使用的是新的额API boolean useNewApi = job.getUseNewReducer(); initialize(job, getJobID(), reporter, useNewApi); // check if it is a cleanupJobTask //和map任务一样,Task有4种,Job-setup Task, Job-cleanup Task, Task-cleanup Task和ReduceTask if (jobCleanup) { //这里执行的是Job-cleanup Task runJobCleanupTask(umbilical, reporter); return; } if (jobSetup) { //这里执行的是Job-setup Task runJobSetupTask(umbilical, reporter); return; } if (taskCleanup) { //这里执行的是Task-cleanup Task runTaskCleanupTask(umbilical, reporter); return; } /* 后面的内容就是开始执行Reduce的Task */ // Initialize the codec codec = initCodec(); boolean isLocal = "local".equals(job.get("mapred.job.tracker", "local")); if (!isLocal) { reduceCopier = new ReduceCopier(umbilical, job, reporter); if (!reduceCopier.fetchOutputs()) { ......到了reduceCopier.fetchOutps()这里必须停一步了,因为后面的Shuffle阶段和Merge阶段都在这里实现:
/** * 开启n个线程远程拷贝Map中的输出数据 * @return * @throws IOException */ public boolean fetchOutputs() throws IOException { int totalFailures = 0; int numInFlight = 0, numCopied = 0; DecimalFormat mbpsFormat = new DecimalFormat("0.00"); final Progress copyPhase = reduceTask.getProgress().phase(); //单独的线程用于对本地磁盘的文件进行定期的合并 LocalFSMerger localFSMergerThread = null; //单独的线程用于对内存上的文件进行进行定期的合并 InMemFSMergeThread inMemFSMergeThread = null; GetMapEventsThread getMapEventsThread = null; for (int i = 0; i < numMaps; i++) { copyPhase.addPhase(); // add sub-phase per file } //建立拷贝线程列表容器 copiers = new ArrayList<MapOutputCopier>(numCopiers); // start all the copying threads for (int i=0; i < numCopiers; i++) { //新建拷贝线程,逐一开启拷贝线程 MapOutputCopier copier = new MapOutputCopier(conf, reporter, reduceTask.getJobTokenSecret()); copiers.add(copier); //添加到列表容器中,并开启此线程 copier.start(); } //start the on-disk-merge thread localFSMergerThread = new LocalFSMerger((LocalFileSystem)localFileSys); //start the in memory merger thread inMemFSMergeThread = new InMemFSMergeThread(); //定期合并的2个线程也开启,也就是说copy阶段和merge阶段是并行操作的 localFSMergerThread.start(); inMemFSMergeThread.start(); // start the map events thread getMapEventsThread = new GetMapEventsThread(); getMapEventsThread.start(); .....在上面的代码中出现很多陌生的Thread的定义,这个可以先不用管,我们发现getMapEventsThread就是在这里开启的,去获取了最新的位置,位置获取完成当然是要启动很多的拷贝线程了,这里叫做MapOutputCopier线程,作者是把他放入一个线程列表中,逐个开启。看看里面的具体实现,他是如何进行远程拷贝的呢。
@Override public void run() { while (true) { try { MapOutputLocation loc = null; long size = -1; synchronized (scheduledCopies) { //从scheduledCopies列表中获取获取map Task的输出数据的位置 while (scheduledCopies.isEmpty()) { //如果scheduledCopies我空,则等待 scheduledCopies.wait(); } //获取列表中的第一个数据作为拷贝的地址 loc = scheduledCopies.remove(0); } CopyOutputErrorType error = CopyOutputErrorType.OTHER_ERROR; readError = false; try { shuffleClientMetrics.threadBusy(); //标记当前的map输出位置为loc start(loc); //进行只要的copy操作,返回拷贝字节数的大小 size = copyOutput(loc); shuffleClientMetrics.successFetch(); //如果进行到这里,说明拷贝成功吗,标记此error的标记为NO_ERROR error = CopyOutputErrorType.NO_ERROR; } catch (IOException e) { //抛出异常,做异常处理 ....从location列表中去取出,然后进行拷贝操作,核心方法在copyOutput(),接着往里跟踪:
..... // Copy the map output //根据loc Map任务的数据输出位置,进行RPC的拷贝 MapOutput mapOutput = getMapOutput(loc, tmpMapOutput, reduceId.getTaskID().getId());继续往里:
private MapOutput getMapOutput(MapOutputLocation mapOutputLoc, Path filename, int reduce) throws IOException, InterruptedException { // Connect //打开url资源定位符的连接 URL url = mapOutputLoc.getOutputLocation(); URLConnection connection = url.openConnection(); //得到远程数据的输入流 InputStream input = setupSecureConnection(mapOutputLoc, connection); ...... //We will put a file in memory if it meets certain criteria: //1. The size of the (decompressed) file should be less than 25% of // the total inmem fs //2. There is space available in the inmem fs // Check if this map-output can be saved in-memory //向ShuffleRamManager申请内存存放拷贝的数据,判断内存是否内存是否装得下,装不下则放入DISK磁盘 boolean shuffleInMemory = ramManager.canFitInMemory(decompressedLength); // Shuffle MapOutput mapOutput = null; if (shuffleInMemory) { if (LOG.isDebugEnabled()) { LOG.debug("Shuffling " + decompressedLength + " bytes (" + compressedLength + " raw bytes) " + "into RAM from " + mapOutputLoc.getTaskAttemptId()); } //如果内存装得下,则将输入流中的数据放入内存 mapOutput = shuffleInMemory(mapOutputLoc, connection, input, (int)decompressedLength, (int)compressedLength); } else { if (LOG.isDebugEnabled()) { LOG.debug("Shuffling " + decompressedLength + " bytes (" + compressedLength + " raw bytes) " + "into Local-FS from " + mapOutputLoc.getTaskAttemptId()); } //装不下,则放入文件中 mapOutput = shuffleToDisk(mapOutputLoc, input, filename, compressedLength); } return mapOutput; }在这里我们看到了,Hadoop通过URL资源定位符,获取远程输入流,进行操作的,在拷贝到本地的时候,还分了2种情况处理,当当前的内存能方得下当前数据的时候,放入内存中,放不下则写入到磁盘中。这里还出现了ShuffleRamManager的用法。至此,Shuffle阶段宣告完成。还是比较深的,一层,又一层的。
Merger阶段。Merge阶段其实是和Shuffle阶段并行进行的,刚刚也看到了,在fetchOutputs中,这些相关进程都是同时开启的,
public boolean fetchOutputs() throws IOException { int totalFailures = 0; int numInFlight = 0, numCopied = 0; DecimalFormat mbpsFormat = new DecimalFormat("0.00"); final Progress copyPhase = reduceTask.getProgress().phase(); //单独的线程用于对本地磁盘的文件进行定期的合并 LocalFSMerger localFSMergerThread = null; //单独的线程用于对内存上的文件进行进行定期的合并 InMemFSMergeThread inMemFSMergeThread = null; ....Merge的主要工作就是合并数据,当内存中或者磁盘中的文件比较多的时候,将小文件进行合并变成大文件。挑出其中的一个run方法
.... public void run() { LOG.info(reduceTask.getTaskID() + " Thread started: " + getName()); try { boolean exit = false; do { exit = ramManager.waitForDataToMerge(); if (!exit) { //进行内存merger操作 doInMemMerge();目的非常明确,就是Merge操作,这是内存文件的合并线程的run方法,LocalFSMerger与此类似,不分析了。这个Mergr处理是并与Shuffle阶段的。在这里这2个阶段都完成了。还是有点复杂的。下面是相关的一些类关系图,主要要搞清4个线程是什么作用的。
4个线程的调用都是在ReduceCopier.fetchOutput()方法中进行的。在Shuffle,Merge阶段的后面就来到了,Sort阶段。
Sort阶段,的任务和轻松,就是完成一次对内存和磁盘总的一次Merge合并操作,其中还会对其中进行一次sort排序操作。
.... //标识copy操作已经完成 copyPhase.complete(); // copy is already complete setPhase(TaskStatus.Phase.SORT); statusUpdate(umbilical); //进行内存和磁盘中的总的merge阶段的操作,Sort包含其中执行 final FileSystem rfs = FileSystem.getLocal(job).getRaw(); RawKeyValueIterator rIter = isLocal ? Merger.merge(job, rfs, job.getMapOutputKeyClass(), job.getMapOutputValueClass(), codec, getMapFiles(rfs, true), !conf.getKeepFailedTaskFiles(), job.getInt("io.sort.factor", 100), new Path(getTaskID().toString()), job.getOutputKeyComparator(), reporter, spilledRecordsCounter, null) : reduceCopier.createKVIterator(job, rfs, reporter);那么Sort操作在哪里呢,就在最下面的createKVIterator中:
private RawKeyValueIterator createKVIterator( JobConf job, FileSystem fs, Reporter reporter) throws IOException { ..... //在Merge阶段对所有的数据进行归并排序 Collections.sort(diskSegments, new Comparator<Segment<K,V>>() { public int compare(Segment<K, V> o1, Segment<K, V> o2) { if (o1.getLength() == o2.getLength()) { return 0; } return o1.getLength() < o2.getLength() ? -1 : 1; } }); // build final list of segments from merged backed by disk + in-mem List<Segment<K,V>> finalSegments = new ArrayList<Segment<K,V>>();,Sort阶段的任务就是这么简单。下面看一下前3个阶段主要的执行流程,这3个阶段构成了Reduce Task的核心。
Reduce阶段,跟随这个图的执行方向,接下来我们应该执行的是key-value的reduce()函数了,没错就是循环键值对,执行此函数
.... //判断执行的是新的API还是旧的API if (useNewApi) { runNewReducer(job, umbilical, reporter, rIter, comparator, keyClass, valueClass); } else { runOldReducer(job, umbilical, reporter, rIter, comparator, keyClass, valueClass); }在这里我们执行的就是runReducer方法了,我们往老的API跳:
private <INKEY,INVALUE,OUTKEY,OUTVALUE> void runOldReducer(JobConf job, TaskUmbilicalProtocol umbilical, final TaskReporter reporter, RawKeyValueIterator rIter, RawComparator<INKEY> comparator, Class<INKEY> keyClass, Class<INVALUE> valueClass) throws IOException { Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE> reducer = ReflectionUtils.newInstance(job.getReducerClass(), job); // make output collector String finalName = getOutputName(getPartition()); //获取输出的key,value final RecordWriter<OUTKEY, OUTVALUE> out = new OldTrackingRecordWriter<OUTKEY, OUTVALUE>( reduceOutputCounter, job, reporter, finalName); OutputCollector<OUTKEY,OUTVALUE> collector = new OutputCollector<OUTKEY,OUTVALUE>() { public void collect(OUTKEY key, OUTVALUE value) throws IOException { //将处理后的key,value写入输出流中,最后写入HDFS作为最终结果 out.write(key, value); // indicate that progress update needs to be sent reporter.progress(); } }; // apply reduce function try { //increment processed counter only if skipping feature is enabled boolean incrProcCount = SkipBadRecords.getReducerMaxSkipGroups(job)>0 && SkipBadRecords.getAutoIncrReducerProcCount(job); //判断是否为跳过错误记录模式 ReduceValuesIterator<INKEY,INVALUE> values = isSkipping() ? new SkippingReduceValuesIterator<INKEY,INVALUE>(rIter, comparator, keyClass, valueClass, job, reporter, umbilical) : new ReduceValuesIterator<INKEY,INVALUE>(rIter, job.getOutputValueGroupingComparator(), keyClass, valueClass, job, reporter); values.informReduceProgress(); while (values.more()) { reduceInputKeyCounter.increment(1); //Record迭代器中获取每一对,执行用户定义的Reduce函数,此阶段为Reduce阶段 reducer.reduce(values.getKey(), values, collector, reporter); if(incrProcCount) { reporter.incrCounter(SkipBadRecords.COUNTER_GROUP, SkipBadRecords.COUNTER_REDUCE_PROCESSED_GROUPS, 1); } //获取下一个key,value values.nextKey(); values.informReduceProgress(); } //...和Map Task的过程很类似,也正如我们预期的那样,循环迭代执行,这就是Reduce阶段。
Write阶段。Write阶段是最后一个阶段,在用户自定义的reduce中,一般用户都会调用collect.collect方法,这时候就是写入的操作了。这时的写入就是将最后的结果写入HDFS作为最终结果了。这里先定义了OutputCollector的collect方法:
OutputCollector<OUTKEY,OUTVALUE> collector = new OutputCollector<OUTKEY,OUTVALUE>() { public void collect(OUTKEY key, OUTVALUE value) throws IOException { //将处理后的key,value写入输出流中,最后写入HDFS作为最终结果 out.write(key, value); // indicate that progress update needs to be sent reporter.progress(); } };至此,完成了Reduce任务的所有阶段。下面是一张时序图,便于理解:
掌握了Map ,Reduce2个过程核心实现的过程将会帮助我们更加理解Hadoop作业运行的整个流程。整个分析的过程也许会有点枯燥,但是苦中作乐。
Reduce Task的学习笔记