Hadoop权威指南读书笔记(4) - MapReducer工作
本文对应原书中第七章。其中的知识对于进一步调优MapReduce非常重要。我们先介绍MapReduce作业的运行过程，再介绍它的容错机制，然后会介绍其中的Shffle和Sort过程。
MapReduce工作过程
我们先用一张图来对整个过程有一个整体把握：
这里面有5个组件:
- client：提交MR作业
- Yarn Resource Manager：协同调度集群的全部资源
- Yarn Node Manager：启动和监视集群中的容器
- MapReduce Application Master：协同MR的tasks。它们都运行在由resource manager分配然后由node manager管理的容器中。
- 分布式文件系统
一个MR作业由如下几个过程组成：
作业提交
Job实例调用submit()方法提交一个作业，这个方法会创建一个JobSubmitter实例(图中第1步)。提交完作业后，waitForCompletion(如果代码中调用了的话)方法会每隔一秒查看作业的状态并且将变化(如果有的话)打印到终端。
在JobSubmitter中会做如下事情：
- 向Yarn RM中请求一个新的应用ID，分配给这个MR作业(第2步)。
- 检查作业的输出需求。例如，如果输出目录已经存在，则不提交作业并且抛出一个异常。
- 计算作业的输入切片(准确说是切片信息)。如果切片不能被计算出来(例如输入目录不存在)，同样作业不会被提交并且发出一个异常。
- 拷贝各种作业资源(例如作业的jar包，配置文件，切片信息等)到分布式文件系统下的一个被命名为job ID的目录下(第3步)。
- 调用submitApplicatiion方法提交作业到Yarn RM(第4步)。
作业初始化
当Yarn RM收到了submitApplicatiion方法中的请求。它将这个请求交给Yarn调度器。调度器会分配一个容器，然后Yarn RM在这个容器中启动一个AM(在NM管理下)(5a和5b)。
AM是一个java应用，它的主类叫做MRAppMaster。这个类会创建一些记录对象，用来获取来自任务的报告(第6步)。然后会从分布式文件系统中获取切片信息(第7步)。为每一个分片创建一个map任务，而reduce任务的个数由mapreduce.job.reduces属性或者setNumReduceTasks方法指定。同样任务的ID也在这个时候被分配。
AM也需要决定如何运行这些任务。如果作业太小，AM则把所有的任务跑在一个JVM中。这种作业被称为uber任务。
任务分配
如果作业不是uber任务，AM则会从RM那里为每一个map和reduce任务请求一个容器(第8步)。当然为map请求容器的优先级要高于reduce。事实上，reduce任务的请求直到至少5%的map任务完成才会被提出。
还需要注意到, map任务会尽量遵守locality约束，而reduce任务则可以运行在集群的任意地方。
任务执行
一旦RM给某个任务分配了某个特定节点上的容器，AM通过NM启动这个容器(9a和9b)。这个任务是一个Java应用并且主类叫做YarnChild。在运行这个任务之前，它会先从分布式文件系统中获取各种资源(第10步)。最后，它运行这个任务(第11步)。
状态更新
当任务运行时，它记录了自己的进度。对于map任务，是自己处理的分片的比例。对于reduce任务，稍微复杂一点，需要把整个过程分为三个部分，即shuffle和sort的三个阶段：copy, sort 和reduce，然后再估算。
每个任务，会每三秒通过一个脐带(umbilical)接口向AM报告自己的状态，而client则每一秒从AM那里收到最新的进度。
作业完成
当AM收到最后一个任务完成的消息时，它会把作业的状态变为“成功“。Job实例则获得这个状态后，告诉用户并且它的waitForCompletion方法会返回值，同时各种统计信息会被打印出来。
稍微提一点的是，通过配置mapreduce.job.end-notification.url property参数，AM也会发送HTTP通知。
最后AM和任务的容器会清理它们的状态。以及一些其他的收尾工作也在这时候被完成。
容错
我们使用MapReduce计算模型的一个重要原因是它自带了容错机制。下面我们把MR中遇到的错误分为如下四类并介绍它们的容错机制：
任务失败
任务失败一般来说也分为以下几种：
- map或者reduce代码中抛出runtime异常，此时 JVM会在退出前向AM报告错误。然后AM会把这个任务标记成失败(failed)，并释放容器使得其他任务能使用它。
- JVM异常退出，此时NM会注意到这个程序的退出并将其报告给AM。
- 如果一个任务超过一定时间(默认10分钟，也可以用mapreduce.task.timeout配置)没有汇报状态，AM会把它标记为失败，并且JVM会将这个任务杀死。如果用户将这个时间设为0，系统会关闭这个功能，这也意味着无论一个任务运行多久，都不会被杀掉，我们应当尽量避免这个情况的出现。
当AM发现一个任务是失败的状态时，它会重新分配这个任务。AM会尝试将任务分配到其他节点上。而当一个任务失败四次时，它将不会在被重新分配。事实上，如果一个任务失败四次，那么整个作业将会失败。当然，有时候用户希望尽管一些任务失败，但仍然要得到(部分)结果。这时可以通过配置mapreduce.map.failures.maxpercent和mapreduce.reduce.failures.maxpercent这两个参数来使得作业允许一部分任务失败。
除了失败以外，任务还有一个状态叫做杀死(killed)。例如，任务在执行时它所在的NM挂掉或者用户通过WebUI手动把它杀掉等，都会造成任务被标记为杀死。杀死的任务是不算在这个任务尝试的次数中的，因为这次失败并不是由于任务本身造成的。
AM失败
类似于上面的任务，AM失败也会有重启机制。但需要注意到的是默认尝试次数是2次，可以用mapreduce.am.max-attempts来设置，即超过2次失败则整个作业就会失败。同时由于Yarn对于它上面运行的的任何Yarn AM默认尝试次数也是2。所以当我们想增加AM尝试次数时，除了修改上面的参数，我们还需要修改yarn.resourcemanager.am.max-attempts这个参数。
由于AM会定期给RM发送报心跳，因此RM可以检测到AM是否失败。如果失败，RM会重启一个AM。并且这个新的AM会通过读取job history来避免重跑已经成功的task。当然这个避免重跑也可以被取消，只需要把yarn.app.mapreduce.am.job.recovery.enable设定为false即可。
另一方面，client在作业初始化时，会从RM那里得到AM的地址并不断从AM那里获取作业状态(比如打印到终端上)。因此当client向AM的请求超时后，它会重新向RM获取新的AM的地址。整个过程是不会让用户察觉到。
NM失败
当一个NM出错时，它向RM发送的心跳就会出现异常频率甚至停止发送。如果RM发现一个NM超过10分钟(可以用yarn.resourcemanager.nm.liveness-monitor.expiry-interval-ms配置)没有发送心跳，那么就会把这个NM从调度器的容器里面移除。
同时这个NM上运行的AM或者任务会像之前两个小节说的那样恢复。同时需要注意到的是，和AM不同，新的NM会重新允许在原来失败的NM上已经运行成功的map任务，因为这些任务的输出可能已经被破坏而无法被后面的reduce任务获取。
如果一个NM对于这个作业失败的次数太高，那么它会被AM加到黑名单中。事实上，对于一个作业，如果超过3个(可以用mapreduce.job.maxtaskfailures.per.tracker配置)任务在一个NM上失败，那么AM会尽量将其分配到其他节点上。更需要注意到的是：目前为止，RM上并没有黑名单机制。因此即使一个NM已经被之前作业的AM加到黑名单中，RM仍然有可能把新的作业分配到这里。
RM失败
RM失败是非常严重的问题，因为这意味着整个Yarn集群不能运作。默认情况下，RM是单点失败的。但现在为了做到HA，很多集群都会设置主备RM。事实上，作业的信息会被存在HA的存储系统里面(例如hdfs zookeeper)，但NM的信息不会被备份(因为NM会给新的RM发送心跳因此很容易重构)，同样任务的信息也不会被存储(因为它们被AM管理)。
当RM重启时，它会从备份中读取AM信息并重新启动全部AM。而NM和client则需要配置对于RM的容错机制。事实上，它们会采取轮询直到找到一个活跃的RM。
Shuffle和Sort
在MR中有2个非常重要的过程:Sort和Shuffle。前者指每一个送到reducer的输入都是有序的(按照key排序)，后者指mapper的输出要被送到reducer当做输入。
下图揭示了这个过程：
2.
Map任务
当map任务开始输出数据时，并不是简单地写到磁盘上。事实上，每个mapper都有一个环形内存缓冲，这个缓存的大小默认是100MB(可以用mapreduce.task.io.sort.mb配置)。当缓冲区的内容超过容量的80%(可以用mapreduce.map.sort.spill.percent配置)时，一个后台线程则会把这些内容写到磁盘上，这些写到磁盘的内容被称为spills。注意写到磁盘和缓冲区是同时进行的，除非缓冲区被写满。
在写到磁盘之前，线程首先把它们按照要被送到的reducer分成多个分区。在每个分区内部，都会执行一个内排序，并且如果应用中有combiner，此时还会执行在排序后的结果上进行combine。
当mapper处理完最后一条数据后，硬盘里面会有多个spill。所以它们还会被合并成一个新的文件，这个文件仍然是分区的且每个分区内部是有序的。此外，如果有至少3个spill，此时还会执行combier函数(如果有的话)。
最后要提的一点是，这些数据都默认是不压缩的，当然你可以参照前的Hadoop I/O中的内容对其压缩。
Reduce任务
我们现在介绍reduce过程。如前面所说，reducer被分为三个过程: Copy, Sort和Reduce。
Copy阶段，reducer需要从多个mapper中的输出文件的某个分区拿到自己想要的数据。因此只要一个mapper完成了，那么reducer就会从它那复制数据。默认是有5个线程抓取mapper的数据，可以通过mapreduce.reduce.shuffle.parallelcopies配置。
Reducer是如何知道自己要去哪取数据的？实际上，mapper会通过心跳告诉AM他们的状况，而reducer则定期去向AM请求数据。
这里需要注意到的是，mapper并不会在reducer抓取到数据后就把他们删除，因为这些reducer有可能失败。实际上AM会告诉他们删除数据的时间，一般是在作业结束的时候。
如果mapper的数据比较小，则会直接被放到JVM的内存中，否则会被放到磁盘上。当内存的缓冲达到一个阈值，它们会被写到磁盘上，并且此时会调用combiner函数。
Sort阶段，当mapper的全部数据都被拷贝过来后，就会进入这个阶段。这个阶段会把所有的数据都合并成一个大的有序的文件。举一个例子，如果有50个mapper的数据，并且合并因子是10(默认值，可以用mapreduce.task.io.sort.factor property配置，类似于mapper)，那么它们会被合并5次并生成一个大文件。注意到如下图所示，这个合并过程总会尽量使得最后一轮合并的文件数目达到这个合并因子，这么做并不会减少合并次数，但是可以减少中间写到磁盘上的文件。
最后则是reduce阶段，这个阶段没有太多要讲的。就是按照key把他们合并起来，并最后写到输出目录里面。
