Map/Reduce hadoop 细节

4、Reduce任务的分配与执行
比之Map任务，Reduce的分配及其简单，基本上是所有Map任务完成了，有空闲的任务服务器，来了就给分配一个Job任务。因为Map任务的结果星罗棋布，且变化多端，真要搞一个全局优化的算法，绝对是得不偿失。而Reduce任务的执行进程的构造和分配流程，与Map基本完全的一致，没有啥可说的了。。。

但其实，Reduce任务与Map任务的最大不同，是Map任务的文件都在本地隔着，而Reduce任务需要到处采集。这个流程是作业服务器经由此Reduce任务所处的任务服务器，告诉Reduce任务正在执行的进程，它需要的Map任务执行过的服务器地址，此Reduce任务服务器会于原Map任务服务器联系（当然本地就免了...），通过FTP服务，下载过来。这个隐含的直接数据联系，就是执行Reduce任务与执行Map任务最大的不同了。

5、作业的完成
当所有Reduce任务都完成了，所需数据都写到了分布式文件系统上，整个作业才正式完成了。此中，涉及到很多的类，很多的文件，很多的服务器，所以说起来很费劲，话说，一图解千语，说了那么多，我还是画两幅图，彻底表达一下吧。。。

首先，是一个时序图。它模拟了一个由3个Map任务和1个Reduce任务构成的作业执行流程。我们可以看到，在执行的过程中，只要有人太慢，或者失败，就会增加一次尝试，以此换取最快的执行总时间。一旦所有Map任务完成，Reduce开始运作（其实，不一定要这样的...），对于每一个Map任务来说，只有执行到Reduce任务把它上面的数据下载完成，才算成功，否则，都是失败，需要重新进行尝试。。。

而第二副图，不是我画的，就不转载了，参见这里，它描述了整个Map/Reduce的服务器状况图，包括整体流程、所处服务器进程、输入输出等，看清楚这幅图，对Map/Reduce的基本流程应该能完全跑通了。有这几点，可能图中描述的不够清晰需要提及一下，一个是在HDFS中，其实还有日志文件，图中没有标明；另一个是步骤5，其实是由TaskTracker主动去拉取而不是JobTracker推送过来的；还有步骤8和步骤11，创建出来的MapTask和ReduceTask，在Hadoop中都是运行在独立的进程上的。。。

IV. Map任务详请
从上面，可以了解到整个Map和Reduce任务的整体流程，而后面要啰嗦的，是具体执行中的细节。Map任务的输入，是分布式文件系统上的，包含键值对信息的文件。为了给每一个Map任务指定输入，我们需要掌握文件格式把它分切成块，并从每一块中分离出键值信息。在HDFS中，输入的文件格式，是由InputFormat<K, V>类来表示的，在JobConf中，它的默认值是TextInputFormat类（见getInputFormat），此类是特化的FileInputFormat<LongWritable, Text>子类，而FileInputFormat<K, V>正是InputFormat<K, V>的子类。通过这样的关系我们可以很容易的理解，默认的文件格式是文本文件，且键是LongWritable类型（整形数），值是Text类型（字符串）。仅仅知道文件类型是不够的，我们还需要将文件中的每一条数据，分离成键值对，这个工作，是RecordReader<K, V>来做的。在TextInputFormat的getRecordReader方法中我们可以看到，与TextInputFormat默认配套使用的，是LineRecordReader类，是特化的RecordReader<LongWritable, Text>的子类，它将每一行作为一个记录，起始的位置作为键，整行的字符串作为值。有了格式，分出了键值，还需要切开分给每一个Map任务。每一个Map任务的输入用InputSplit接口表示，对于一个文件输入而言，其实现是FileSplit，它包含着文件名、起始位置、长度和存储它的一组服务器地址。

当Map任务拿到所属的InputSplit后，就开始一条条读取记录，并调用用于定义的Mapper，进行计算（参见MapRunner<K1, V1, K2, V2>和MapTask的run方法），然后，输出。MapTask会传递给Mapper一个OutputCollector<K, V>对象，作为输出的数据结构。它定义了一个collect的函数，接受一个键值对。在MapTask中，定义了两个OutputCollector的子类，一个是MapTask.DirectMapOutputCollector<K, V>，人如其名，它的实现确实很Direct，直截了当。它会利用一个RecordWriter<K, V>对象，collect一调用，就直接调用RecordWriter<K, V>的write方法，写入本地的文件中去。如果觉着RecordWriter<K, V>出现的很突兀，那么看看上一段提到的RecordReader<K, V>，基本上，数据结构都是对应着的，一个是输入一个是输出。输出很对称也是由RecordWriter<K, V>和OutputFormat<K, V>来协同完成的，其默认实现是LineRecordWriter<K, V>和TextOutputFormat<K, V>，多么的眼熟啊。

除了这个非常直接的实现之外，MapTask中还有一个复杂的多的实现，是MapTask.MapOutputBuffer<K extends Object, V extends Object>。有道是简单压倒一切，那为什么有很简单的实现，要琢磨一个复杂的呢。原因在于，看上去很美的往往带着刺，简单的输出实现，每调用一次collect就写一次文件，频繁的硬盘操作很有可能导致此方案的低效。为了解决这个问题，这就有了这个复杂版本，它先开好一段内存做缓存，然后制定一个比例做阈值，开一个线程监控此缓存。collect来的内容，先写到缓存中，当监控线程发现缓存的内容比例超过阈值，挂起所有写入操作，建一个新的文件，把缓存的内容批量刷到此文件中去，清空缓存，重新开放，接受继续collect。

为什么说是刷到文件中去呢。因为这不是一个简单的照本宣科简单复制的过程，在写入之前，会先将缓存中的内存，经过排序、合并器（Combiner）统计之后，才会写入。如果你觉得Combiner这个名词听着太陌生，那么考虑一下Reducer，Combiner也就是一个Reducer类，通过JobConf的setCombinerClass进行设置，在常用的配置中，Combiner往往就是用用户为Reduce任务定义的那个Reducer子类。只不过，Combiner只是服务的范围更小一些而已，它在Map任务执行的服务器本地，依照Map处理过的那一小部分数据，先做一次Reduce操作，这样，可以压缩需要传输内容的大小，提高速度。每一次刷缓存，都会开一个新的文件，等此任务所有的输入都处理完成后，就有了若干个有序的、经过合并的输出文件。系统会将这些文件搞在一起，再做一个多路的归并外排，同时使用合并器进行合并，最终，得到了唯一的、有序的、经过合并的中间文件（注：文件数量等同于分类数量，在不考虑分类的时候，简单的视为一个...）。它，就是Reduce任务梦寐以求的输入文件。

除了做合并，复杂版本的OutputCollector，还具有分类的功能。分类，是通过Partitioner<K2, V2>来定义的，默认实现是HashPartitioner<K2, V2>，作业提交者可以通过JobConf的setPartitionerClass来自定义。分类的含义是什么呢，简单的说，就是将Map任务的输出，划分到若干个文件中（通常与Reduce任务数目相等），使得每一个Reduce任务，可以处理某一类文件。这样的好处是大大的，举一个例子说明一下。比如有一个作业是进行单词统计的，其Map任务的中间结果应该是以单词为键，以单词数量为值的文件。如果这时候只有一个Reduce任务，那还好说，从全部的Map任务那里收集文件过来，分别统计得到最后的输出文件就好。但是，如果单Reduce任务无法承载此负载或效率太低，就需要多个Reduce任务并行执行。此时，再沿用之前的模式就有了问题。每个Reduce任务从一部分Map任务那里获得输入文件，但最终的输出结果并不正确，因为同一个单词可能在不同的Reduce任务那里都有统计，需要想方法把它们统计在一起才能获得最后结果，这样就没有将Map/Reduce的作用完全发挥出来。这时候，就需要用到分类。如果此时有两个Reduce任务，那么将输出分成两类，一类存放字母表排序较高的单词，一类存放字母表排序低的单词，每一个Reduce任务从所有的Map任务那里获取一类的中间文件，得到自己的输出结果。最终的结果，只需要把各个Reduce任务输出的，拼接在一起就可以了。本质上，这就是将Reduce任务的输入，由垂直分割，变成了水平分割。Partitioner的作用，正是接受一个键值，返回一个分类的序号。它会在从缓存刷到文件之前做这个工作，其实只是多了一个文件名的选择而已，别的逻辑都不需要变化。。。

除了缓存、合并、分类等附加工作之外，复杂版本的OutputCollector还支持错误数据的跳过功能，在后面分布式将排错的时候，还会提及，标记一下，按下不表。。。

V. Reduce任务详情
理论上看，Reduce任务的整个执行流程要比Map任务更为的罗嗦一些，因为，它需要收集输入文件，然后才能进行处理。Reduce任务，主要有这么三个步骤：Copy、Sort、Reduce（参见ReduceTask的run方法）。所谓Copy，就是从执行各个Map任务的服务器那里，收罗到本地来。拷贝的任务，是由ReduceTask.ReduceCopier类来负责，它有一个内嵌类，叫MapOutputCopier，它会在一个单独的线程内，负责某个Map任务服务器上文件的拷贝工作。远程拷贝过来的内容（当然也可以是本地了...），作为MapOutput对象存在，它可以在内存中也可以序列化在磁盘上，这个根据内存使用状况来自动调节。整个拷贝过程是一个动态的过程，也就是说它不是一次给好所有输入信息就不再变化了。它会不停的调用TaskUmbilicalProtocol协议的getMapCompletionEvents方法，向其父TaskTracker询问此作业个Map任务的完成状况（TaskTracker要向JobTracker询问后再转告给它...）。当获取到相关Map任务执行服务器的信息后，都会有一个线程开启，做具体的拷贝工作。同时，还有一个内存Merger线程和一个文件Merger线程在同步工作，它们将新鲜下载过来的文件（可能在内存中，简单的统称为文件...），做着归并排序，以此，节约时间，降低输入文件的数量，为后续的排序工作减负。

Sort，排序工作，就相当于上述排序工作的一个延续。它会在所有的文件都拷贝完毕后进行，因为虽然同步有做着归并的工作，但可能留着尾巴，没做彻底。经过这一个流程，该彻底的都彻底了，一个崭新的、合并了所有所需Map任务输出文件的新文件，诞生了。而那些千行万苦从其他各个服务器网罗过来的Map任务输出文件，很快的结束了它们的历史使命，被扫地出门一扫而光，全部删除了。

所谓好戏在后头，Reduce任务的最后一个阶段，正是Reduce本身。它也会准备一个OutputCollector收集输出，与MapTask不同，这个OutputCollector更为简单，仅仅是打开一个RecordWriter，collect一次，write一次。最大的不同在于，这次传入RecordWriter的文件系统，基本都是分布式文件系统，或者说是HDFS。而在输入方面，ReduceTask会从JobConf那里调用一堆getMapOutputKeyClass、getMapOutputValueClass、getOutputKeyComparator等等之类的自定义类，构造出Reducer所需的键类型，和值的迭代类型Iterator（一个键到了这里一般是对应一组值）。具体实现颇为拐弯抹角，建议看一下Merger.MergeQueue，RawKeyValueIterator，ReduceTask.ReduceValuesIterator等等之类的实现。有了输入，有了输出，不断循环调用自定义的Reducer，最终，Reduce阶段完成。。。

VI. 分布式支持
1、服务器正确性保证
Hadoop Map/Reduce服务器状况和HDFS很类似，由此可知，救死扶伤的方法也是大同小异。废话不多说了，直接切正题。同作为客户端，Map /Reduce的客户端只是将作业提交，就开始搬个板凳看戏，没有占茅坑的行动。因此，一旦它挂了，也就挂了，不伤大雅。而任务服务器，也需要随时与作业服务器保持心跳联系，一旦有了问题，作业服务器可以将其上运行的任务，移交给它人完成。作业服务器，作为一个单点，非常类似的是利用还原点（等同于HDFS的镜像）和历史记录（等同于HDFS的日志），来进行恢复。其上，需要持久化用于恢复的内容，包含作业状况、任务状况、各个任务尝试的工作状况等。有了这些内容，再加上任务服务器的动态注册，就算挪了个窝，还是很容易恢复的。JobHistory是历史记录相关的一个静态类，本来，它也就是一个干写日志活的，只是在Hadoop的实现中，对日志的写入做了面向对象的封装，同时又大量用到观察者模式做了些嵌入，使得看起来不是那么直观。本质上，它就是打开若干个日志文件，利用各类接口来往里面写内容。只不过，这些日志，会放在分布式文件系统中，就不需要像HDFS那样，来一个SecondXXX随时候命，由此可见，有巨人在脚下踩着，真好。JobTracker.RecoveryManager类是作业服务器中用于进行恢复相关的事情，当作业服务器启动的时候，会调用其recover方法，恢复日志文件中的内容。其中步骤，注释中写的很清楚，请自行查看。。。

2、任务执行的正确和速度
整个作业流程的执行，秉承着木桶原理。执行的最慢的Map任务和Reduce任务，决定了系统整体执行时间（当然，如果执行时间在整个流程中占比例很小的话，也许就微不足道了...）。因此，尽量加快最慢的任务执行速度，成为提高整体速度关键。所使用的策略，简约而不简单，就是一个任务多次执行。当所有未执行的任务都分配出去了，并且先富起来的那部分任务已经完成了，并还有任务服务器孜孜不倦的索取任务的时候，作业服务器会开始炒剩饭，把那些正在吭哧吭哧在某个服务器上慢慢执行的任务，再把此任务分配到一个新的任务服务器上，同时执行。两个服务器各尽其力，成王败寇，先结束者的结果将被采纳。这样的策略，隐含着一个假设，就是我们相信，输入文件的分割算法是公平的，某个任务执行慢，并不是由于这个任务本身负担太重，而是由于服务器不争气负担太重能力有限或者是即将撒手西去，给它换个新环境，人挪死树挪活事半功倍。

当然，肯定有哽咽的任务，不论是在哪个服务器上，都无法顺利完成。这就说明，此问题不在于服务器上，而是任务本身天资有缺憾。缺憾在何处？每个作业，功能代码都是一样的，别的任务成功了，就是这个任务不成功，很显然，问题出在输入那里。输入中有非法的输入条目，导致程序无法辨识，只能挥泪惜别。说到这里，解决策略也浮出水面了，三十六计走位上，惹不起，还是躲得起的。在MapTask中的MapTask.SkippingRecordReader<K, V>和ReduceTask里的ReduceTask.SkippingReduceValuesIterator<KEY,VALUE>，都是用于干这个事情的。它们的原理很简单，就是在读一条记录前，把当前的位置信息，封装成SortedRanges.Range对象，经由Task的reportNextRecordRange方法提交到TaskTracker上去。TaskTracker会把这些内容，搁在TaskStatus对象中，随着心跳消息，汇报到JobTracker上面。这样，作业服务器就可以随时随刻了解清楚，每个任务正读取在那个位置，一旦出错，再次执行的时候，就在分配的任务信息里面添加一组SortedRanges信息。MapTask或ReduceTask读取的时候，会看一下这些区域，如果当前区域正好处于上述雷区，跳过不读。如此反复，正可谓，道路曲折，前途光明啊。。。

VII. 总结
对于Map/Reduce而言，真正的困难，在于提高其适应能力，打造一款能够包治百病的执行框架。Hadoop已经做得很好了，但只有真正搞清楚了整个流程，你才能帮助它做的更好。。。