【hadoop权威指南第四版】第六章MR的工作原理【笔记+代码】

6.1 运行MR作业

工作原理

四大模块：

客户端，提交MR作业。
jobtracker，协调作业的运行。jobtracker 是一个java应用程序，主类是Jobtracker。
tasktracker，运行作业划分后的任务。tasktracker是一个java应用程序，主类是Tasktracker。
分布式文件系统（一般为HDFS），用来在其他实体间共享作业文件。

6.1.1 提交作业

JobClient的 submitJob()方法所实现的作业提交过程如下。

1.向 jobtracker请求一个新的作业ID(通过调用 JobTracker的
getNewJobId()(步骤2)。
2.检查作业的输出说明。比如，如果没有指定输出目录或者它已经存在，作业就不会被提交，并有错误返回给 MapReduce程序。
3.计算作业的输入划分。如果划分无法计算，比如因为输入路径不存在，作业就不会被提交，并有错误返回给 MapReduce程序。
4.将运行作业所需要的资源—包括作业JAR文件、配置文件和计算所得的输入划分—复制到一个以作业ID号命名的目录中 jobtracker的文件系统。作业JAR的副本较多(由 marred. submit. replication属性控制,默认为10），如此一来，在 tasktracker运行作业任务时，集群能为它们提供许多副本进行访问。（步骤3）。
5.告诉 jobtracker作业准备执行(通过调用 JobTracker的 submitjob()方
法）（步骤4）。

6.1.3 任务的分配

Task Tracker执行一个简单的循环,定期发送心跳( heartbeat)方法调用 Jobtracker心跳方法告诉 jobtracker, tasktracker是否还存活,同时也充当两者之间的消息通道。作为心跳方法调用的一部分， tasktracker会指明它是否已经准备运行新的任务，如果是， jobtracker会为它分配一个任务，并使用心跳方法的返回值与tasktracker进行通信(步骤7)。
针对map任务和 reduce任务, tasktracker有固定数量的槽。例如,一个 asktracker可能可以同时运行两个map任务和两个 reduce任务。(准确数量由 tasktracker核的数量和内存大小来决定，详见第9章的“内存”小节）。默认调度器在处理 reduce任务槽之前，会填满空闲的map任务槽，因此，如果至少有一个空闲的map任务槽, jobtracker会为它选择一个map任务;否则选择一个 reduce任务。
要选择一个 reduce任务, jobtracker只是简单地从尚未运行的 reduce任务列表中选取下一个来执行，并没有考虑数据的本地化。然而，对于一个map任务，它考虑的是 tasktracker的网络位置和选取一个距离其输入划分文件最近的 tasktracker。在最理想的情况下，任务是data- local（数据本地化）的，与分割文件所在节点运行在相同的节点上。同样，任务也可能是rack- local（机架本地化）的：和分割文件在同个机架，但不在同一节点。一些任务既不是数据本地化的，也不是机架本地化的，从与它们自身运行的不同机架上检索数据。可以通过查看作业的计数器得知每种类型任务的比例（详见第8章的“内置计数器”小节）。

6.1.4任务的执行

现在， tasktracker已经被分配了任务，下一步是运行任务。首先，它本地化作业的JAR文件，将它从共享文件系统复制到 tasktracker所在的文件系统。同时，将应用程序所需要的全部文件从分布式缓存复制到到本地磁盘（详见第8章的“分布式缓存”小节）（步骤8）。然后，为任务新建一个本地工作目录，并把JAR文件中的内容解压到这个文件夹下。第三步，新建一个 TaskRunner实例来运行任务。
TaskRunner启动一个新的ava虚拟机（步骤9）来运行每个任务（步骤10），使得用户定义的map和 reduce函数的任何缺陷都不会影响 tasktracker(比如导致它崩溃或者挂起）。但在不同的任务之间重用JVM还是可能的（详见本章的“任务JVM重用”小节）。
子进程通过 umbilical接口与父进程进行通信。它每隔几秒便告知父进程它的进度，直到任务完成。

6.1.5进度的和状态的更新

如果任务报告了进度，便会设置一个标志以表明状态变化将被发送到 tasktracker在另一个线程中，每隔三秒检查此标志一次，如果已设置，则告知 tasktracker当前任务状态。同时, tasktracker每隔五秒发送心跳到 jobtracker(5秒这个间隔是最小值，因为心跳间隔是由集群的大小来决定的：对于一个更大的集群，间隔会更长一些），并且在此调用（指心跳调用）中，所有由tasktracker运行的任务，它们的状态都会被发送至 jobtracker计数器的发送间隔通常大于5秒，因为计数器占的带宽相对较高。

6.2失败

6.2.1任务失败

对于流任务，如果流进程以非零退出代码退出运行，则会被标记为 failed（失败）这种行为是由 stram non.zeroexit.s. failure属性(默认值为true)决定的。
另一种错误情况是子JVM突然退出可能有JVM错误，由 MapReduce用户代码某些特殊原因而造成JVM退出。在这种情况下， tasktracker会注意到进程已经退出，并将此次尝试标记为 failed（失败）
对于任务的挂起，处理方式则有不同。 tasktracker注意到自己已经有一段时间没有收到进度更新，因此进而将任务标记为 failed（失败）。在此之后，子JVM进程将被自动杀死。任务失败的超时间隔通常为10分钟，这可按照每个作业的方式进行设置（或按照每个集群的方式进行设置，具体做法是把 mapred.task. timeout属性设置为一个毫秒为单位的值。
将超时间隔设置为0将关闭超时判定因而造成长时间运行的任务永远不会被标记为 failed。在这种情况下，被挂起的任务永远不会释放它的槽，最终降低整个集群的效率。因此，尽量避免采用这种设置，同时确保任务能够定期汇报其进度。（见本章补充知识“MapReduce的进度如何构成”）
jobtracker被通知一个任务尝试失败时(通过tasktracker的心跳调用实现),它将重新调度该任务的执行。 jobtracker会尝试避免重新调度之前失败过的 tasktracker上的任务。此外，如果一个任务的失败次数超过4次，它将不会再被重试。这个值是可以设置的：尝试运行任务的最多次数，对于map任务，是由 mapredmapmax. attempts属性控制,而对于 reduce任务,则由 mapred. reducemax. attempts属性控制。在默认情况下,如果有任何任务失败次数大于4（或被配置的某个值），整个作业都会失败。
对于一些应用程序，我们不希望一旦有任务失败就中止运行整道作业，因为即使任务失败，其结果可能还是可用的。在这种情况下，可以通过 mapredmax，map. failures. percent和 mapred.max reduce. failures. percent属性来设置在不触发任务失败的情况下允许任务失败的最多次数。
任务尝试也是可以杀死的，这不同于在它失败时被杀死。任务尝试被杀死可能是由于它是一个推测副本（相关详情可参见本章的“推测式执行”），或因为它所处的 tasktracker失败了，这样一来， jobtracker便将在它上面运行的所有任务尝试标记为 killed被杀死的任务尝试不会被计入任务运行尝试次数(由 mapred.map.max. attempts和mapred.reduce.max.attempts设置），因为这不是由于任务失败而导致的尝试禁止。
用户也可以使用网络用户界面或者命令行（输入hadoop job来查看响应的选项）来杀死或取消任务执行尝试。也可以采用前面描述的相同的机制来杀死作业。

6.2.2 tasktracker失败

tasktracker失败是另一种失败形式。如果某 tasktracker由于崩溃或运行过于缓慢而失败,它将停止向 jobtracker发送心跳(或者很少发送心跳)。 jobtracker会注意到此tasktracker已经停止发送心跳（如果它有10分钟仍没有接收到心跳，这由marred. tasktracker. expiry. Interva1属性来设置,以毫秒为单位),会将它从等待任务调度的 tasktracker池中移除。
jobtracker会安排此 tasktracker上已经运行并成功完成的map任务返回，如果它们属于未完成的作业的话，因为它们的中间输出都存放在故障 tasktracker的本地文件系统上，这是 reduce任务无法访问的。
任何进行中的任务也都会被重新调度。
tasktracker还可以被 jobtracker放入黑名单,即使它并没有失败。如果在它上面的
任务的失败次数远远高于集群平均任务失败次数，它就会被放入黑名单。被放入黑名单中的 tasktracker可以通过重启从 jobtracker的黑名单中被移出

6.2.3 jobtracker失败

jobracker失败是在所有失败中最严重的一种。目前， Hadoop没有用于处理jobtracker失败的机制—它是一个单点故障—因此在这种情况工作下，作业注定会失败。然而，这种失败形式发生的概率很小，因为具体某台机器失败的几率很小。未来版本的 Hadoop可能会通过运行多个 jobtracker的方法来解决这个问题，任何时候,都只有其中一个是主 jobtracker使用 ZooKeeper作为 jobtracker的一种协调机制来决定哪一个是主 jobtracker，详见第13章

6.4 shffle和排序

6.4.1 map端

map函数开始产生输出结果时，并不是简单地将它写到磁盘。这个过程更复杂，他利用缓冲的方式写到内存，并处于效率的原因预先进行排序。

每个map任务都有一个环形内存缓冲区，任务会把输出写到此。默认情况下，缓冲区的大小为100MB，此值可以通过io.sort.mb属性来修改。当缓冲内容达到指定大小时（io.sort.spi11. percent，默认为0.80，80%），一个后台线程便开始把内容溢写（spi）到磁盘中。在线程工作的同时，map输出继续被写到缓冲区但如果在此期间缓冲区被填满，map会阻塞直到溢写过程结束。

溢写将按轮询方式写到 marred.loca1.dir属性指定的目录，在一个作业相关子目录中。在写到磁盘之前，线程首先根据数据最终被传送到的 reducer，将数据划分成相应的分区。在每个分区中，后台线程按键进行内排序（in- memory sort）。此时如果有一个combiner，它将基于排序后输出运行。

一旦内存缓冲区达到溢写阈值，就会新建一个溢写文件，因此在map任务写入其最后一个输出记录之后，会有若干个溢写文件。在任务完成之前，溢写文件被合并成一个已分区且已排序的输出文件。配置属性io.sort. factor控制着一次最多能
合并多少流，默认值是10。

如果已经指定 combiner,并且溢写次数至少为3(min.num.spi11s.for. combine属性的值）时， combiner就在输出文件被写之前执行。前面曾讲过， combiner会针对输入反复运行，但不会影响最终结果。运行 combiner的意义在于使map输出更紧凑，从而只有较少数据被写到本地磁盘然后传给 reducer。

map输出被写到磁盘时，对它进行压缩往往是个很好的主意，因为这样会让写入磁盘的速度更快，节约磁盘空间和减少传给 reducer的数据量。默认情况下，输出是不压缩的,但是只要将 marred. compress.map. output设置为true,就可以启用此功能。使用的压缩库由 marred.map. output.compression.codec定义,要想
进一步了解压缩格式，请参见第4章的“压缩”小节。

reducer通过HTTP得到输出文件的分区。用于服务于文件分区的工作线程，其数量由任务的trackerhttpthreads属性来控制。此设置针对的是每个tasktracker，而不是针对每个map任务槽。默认是40，在运行大规模作业的大型集
群上，此值可以根据需要而增加

6.4.2 reduce端

现在转到处理过程的 reduce这一端。map输出文件位于运行map任务的tasktracker的本地磁盘(注意,尽管map输出经常写到 map tasktracker的本地磁盘，但 reduce输出并不这样），不过在现在， tasktracker需要它为分区文件运行reduce任务。而且， reduce任务需要为其特定分区文件从集群上若干个map任务的map输出。map任务可以在不同时间完成，因此只要有一个任务结束， reduce任务就开始复制其输出。这就是 reduce任务的复制阶段。 reduce任务有少量复制线程，因此能够并行地取得map输出。默认是5个线程，但这个默认值可以通过设置 mapped. reduce.para11e1, copies属性来改变

注意: reducer如何知道要从哪个 tasktracker取得map输出呢?

map任务成功完成后,它们会通知其父 tasktracker状态已更新,然后 tasktracker进而通知 jobtracker。这些通知在前面介绍的心跳交流机制中传输。因此，对于指定作业, jobtracker知道map输出和 tasktracker之间的映射关系。 reducer中的一个线程定期向 jobtracker获取map输出位置，直到得到所有输出位置。
tasktracker并没有在第一个 reducer检索之后就立即从磁盘上删除map输出,因为reduce可能失败。反之，他们会等待，直到被jobtracker告知可以删除，这是作业完成后才执行的。

6.4.3配置的调整

map端

reduce端

优化的一般原则是为 shuffle指定尽量多的内存空间。然而，有一个平衡问题，你要确保map和 reduce函数能得到足够的内存来运行。这就是为什么编写map和educe函数时尽量少用内存的原因——它们不应使用不限量的内存（例如，应避免在map中堆积一系列的值）。

为map和 reduce任务运行的JVM指定的内存大小由 marred.chi1a.java.opts
属性来设置。应该让任务节点上这个内存大小尽量大，在第9章的“内存”小节，我们将讨论分配内存中所有需要考虑的约束条件
在map这一端，可以通过避免多次磁盘溢写来获得最佳性能。如果你能估计map输出大小，就可以合理地设置ip，sort，*属性来减少溢写的次数。具体说来，如果可以，应该增加io.sort.mb的值。 MapReduce计数器将计算在作业运行过程中溢写到磁盘中的记录总数，这对于调优很有帮助。注意，计数包括map和 reduce
两端的溢写。

在 reduce这一端，当中间值能够全部存放在内存中时，就能获得最佳性能。默认情况下，这是不可能发生的，因为一般情况下所有内存都预留给 reduce函数。但是如果 reduce函数的内存需求很小,那么将 marred. I nmem. merge. threshold设置为0,将 marred.job, reduce. Input. buffer. percent设置为1.0(或者一个更低的值，详见表6-2）会带来性能的提升。

更常见的情况是， Hadoop使用默认为4KB的缓冲区，这是很低的，因此应该在集群中增加这个值（通过设置io.file. buffer，size，详见第9章的“其他Hadoop属性"小节)。

在2008年4月， Hadoop在通用TB字节排序基准测试中获胜（详见附录A“在
Apache Hadoop中的TB字节排序”），它使用的一个优化方法就是将中间值保存在reduce这一端的内存中。

！Map端的Shuffle

1.MapTask在拿到切片之后，默认会对数据进行按行读取，每读取一行默认调用一次map方法来进行处理
2.每一个MapTask默认自带一个缓冲区，map方法执行的结果会临时的写到缓冲区中
3.缓冲区是维系在内存中，默认是100M
4.当缓冲区达到一定条件的时候，就会将缓冲区中的数据写到磁盘上，这个过程称之为溢写(spill)
5.溢写之后，map方法产生的结果会继续写到缓冲区中
6.多次溢写之后，会产生多个溢写文件，MapTask处理完数据之后，会将所有的溢写文件合并成一个文件，
这个过程称之为合并(merge)，merge过程不会减少数据量。在merge过程中，如果缓冲区中依然有数据，
则将缓冲区中的数据一起合并到最后的结果文件中(final out)
7.如果没有产生溢写，则最后会把缓冲区中的数据直接写出到最后的结果文件中
8.数据写到缓冲区中的时候，数据会在缓冲区中进行分区(partition)、排序(sort)。如果在这个过程中指定了Combiner，
那么数据在缓冲区中还会进行combine操作 - 这也就意味着数据在缓冲区中是分区且有序的 - 也同样意味着单个溢写
文件中的数据也是有序且分区的 - 如果整体来看，所有的溢写文件应该是局部分区并且局部有序的 - 数据在缓冲区中
排序采用的是快速排序
9.在merge过程中，会对文件再次进行分区并且排序，所以最后的结果文件中是分区且有序的 - merge过程中的排序
采用的是归并排序
10.如果指定了Combiner，如果溢写文件的个数>=3个，那么merge的时候会再进行一次combine过程
11.注意问题：
（1）缓冲区本质上是一个字节数组
（2）阈值默认是0.8，即当缓冲区的使用量达到80%的时候，这个时候就会进行溢写
（3）缓冲区中的数据大小不代表溢写文件的大小，即溢写文件的大小不一定是80mb，因为需要序列化
（4）输入的数据量也不决定溢写次数
（5）缓冲区是环形缓冲区，好处在于可以重复利用这个缓冲区而不用重复寻址

！Reduce端的Shuffle

1.ReduceTask会启动fetch线程去Map端抓取数据
2.在抓取数据的时候，会只抓取当前ReduceTask所对应的分区的数据
3.抓取完数据之后，会对数据进行merge，将所有的数据合并到一个文件中，
并且在合并过程中会进行排序 - 采用的排序机制依然是归并排序
4.将相同的键所对应的值放在同一组中，产生一个针对值的迭代器，这个过程称之为分组(Group)
5.分组完成之后，每一个键调用一次reduce方法利用reduce方法来进行处理，最后将处理结果写到HDFS上
6.注意问题：
（1）fetch线程数量默认为5个
（2）merge因子默认为10，表示每10个文件合并成1个文件，然后再次merge，最终合并为一个文件
（3）ReduceTask的阈值默认为0.05，即5%的MapTask结束之后，ReduceTask就会启动开始抓取数据

！Shuffle的调优

1.调大缓冲区，一般情况下会将缓冲区大小设置为250M~400M之间，以减少溢写次数
2.可以增大缓冲区的阈值
3.增加Combine过程
4.在网络资源紧张的情况下，可以考虑将数据进行压缩
5.增多fetch线程的数量，一般的做法是让此线程数接近或等于map task 数量。达到并行抓取的目的。
6.增大merge因子
7.减小ReduceTask的阈值

6.5 任务的执行

6.5.1推测式执行

任务可能有与多种原因而执行缓慢，包括硬件老化或者软件配置错误。hadoop不会尝试诊断或者修复执行缓慢的任务。相反，在一个任务运行比预期慢的时候，他会进行检测，启动另一个相同的任务作为备份。这就是所谓的任务的推测式执行。

当然，推测式执行只有在一个作业中所有任务都启动后才启动。并且推测式执行任务只针对已经运行一段时间（至少一分钟）且比作业中其他任务的平均进度慢的任务。源本和推测式副本中有一个任务成功后，其他都会被终止。

推测式执行是一种优化措施，他并不能是作业的运行更加可靠。例如，运行缓慢是由于BUG造成的，推测执行多少次也是徒劳。

推测式执行目的是减少作业执行时间，但是这是以集群效率为代价的。在一个繁忙的集群中，推测式执行会减少整体的吞吐量，因为冗余的任务在执行时会减慢作业的执行时间。

6.5.2任务JVM重用

为每个任务启动一个新的JVM将消耗大概1秒，对于运行1min的作业而言，这是微不足道的。但是对于有些执行时间非常短的任务来说，这个开销就较大。所以对后续任务重用JVM，可以获得性能的提升。

多个任务在一个JVM中是按顺序运行的。tasktracker可以同时运行多个任务，但都运行在独立的JVM中。

控制任务JVM重用的属性是mapred.job.reuse.jvm.num.tasks。他定义指定作业每个JVM运行的task任务的最大数量。来自不同的任务总是运行在不同的JVM上。蜀国属性为-1，则同一作业共享一个JVM的任务数量不限。JobConf中的setNumTaskToExecutePerJVM（）方法也可以用于设置这个属性。

6.5.3 跳过坏记录

由于数据而导致任务抛出一个异常，那么重新运行任务将无济于事，他每次都会因为相同的原因而失败。

如果错误出现于mapper和reducer，我们可以检测出错误记录并忽略，或抛出异常来取消该作业。但是如果bug出现在第三方的库，我们在mapper和reducer无法修改。所以就可以采用hadoop的skipping模式（忽略模式）来自动跳过错误记录。该模式启动后，任务将正在处理的记录报告给tasktracker。任务失败事，tasktracker将会重新运行该任务，跳过刚刚那个失败的记录。

执行步骤如下：

1、任务失败

2、任务失败

3、开启skipping模式。任务失败但是错误记录仍然储存在tasktracker中。

4、skipping模式仍在启用。任务忽略上一次失败的错误记录继续运行。

该模式只能检测出一个错误记录。
可以设置任务尝试最大值。（mapred.map.max.attemps和mapred.reduce.max.attemps）。
错误录保存在_logs/skip目录下。