Hadoop学习（二）——MapReduce\Yarn架构

其余更多java基础文章：
java基础学习(目录)

---

学习资料
理解Hadoop YARN架构

本文先讲MapReduce 1.x的框架。再讲MapReduce 1.x升级改进后MapReduce 2.x/Yarn的框架。目前主要是用MapReduce 2.x/Yarn的框架。

MapReduce 1.x

MapReduce 1.x重点概念

JobClient

用户编写的MapReduce程序经过Client提交到JobTracker端；同时，用户可经过Client提供的一些接口查看做业运行状态。在Hadoop内部用“做业” （Job）表示MapReduce程序。每个Job都会在用户端经过Client类将应用程序以及参数配置Configuration打包成Jar文件存储在HDFS，并把路径提交到JobTracker，而后由JobTracker建立每个Task（即MapTask和ReduceTask），将它们分发到各个TaskTracker服务中去执行。

JobClient提交Job的详细流程主要以下：

JobClient在获取了JobTracker为Job分配的id以后，会在JobTracker的系统目录(HDFS)下为该Job建立一个单独的目录，目录的名字便是Job的id，该目录下会包含文件job.xml、job.jar、job.split等，其中，job.xml文件记录了Job的详细配置信息，job.jar保存了用户定义的关于job的map、reduce操纵，job.split保存了job任务的切分信息。

JobTracker

JobTracker 主要负责资源监控和做业调度。JobTracker 监控全部 TaskTracker 与做业Job的健康情况，一旦发现失败状况后，其会将相应的任务转移到其余节点；同时，JobTracker 会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器，而调度器会在资源出现空闲时，选择合适的任务使用这些资源。在Hadoop 中，任务调度器是一个可插拔的模块，用户能够根据本身的须要设计相应的调度器。

如下引用 www.aboutyun.com/thread-7778…

JobTracker为做业的提交作了两件事：一.为做业生成一个Job；二.接受该做业。 咱们都知道，客户端的JobClient把做业的全部相关信息都保存到了JobTracker的系统目录下(固然是HDFS了)，这样作的一个最大的好处就是客户端干了它所能干的事情同时也减小了服务器端JobTracker的负载。下面就来看看JobTracker是如何来完成客户端做业的提交的吧！哦。对了，在这里我不得不提的是客户端的JobClient向JobTracker正式提交做业时直传给了它一个改做业的JobId，这是由于与Job相关的全部信息已经存在于JobTracker的系统目录下，JobTracker只要根据JobId就能获得这个Job目录。

对于上面的Job的提交处理流程，我将简单的介绍如下几个过程：

1. 建立Job的JobInProgress
   JobInProgress对象详细的记录了Job的配置信息，以及它的执行状况，确切的来讲应该是Job被分解的map、reduce任务。在JobInProgress对象的建立过程当中，它主要干了两件事，一是把Job的job.xml、job.jar文件从Job目录copy到JobTracker的本地文件系统(job.xml->/jobTracker/jobid.xml，job.jar->/jobTracker/jobid.jar)；二是建立JobStatus和Job的mapTask、reduceTask存队列来跟踪Job的状态信息。
2. 检查客户端是否有权限提交Job
   JobTracker验证客户端是否有权限提交Job其实是交给QueueManager来处理的。
3. 检查当前mapreduce集群可以知足Job的内存需求
   客户端提交做业以前，会根据实际的应用状况配置做业任务的内存需求，同时JobTracker为了提升做业的吞吐量会限制做业任务的内存需求，因此在Job的提交时，JobTracker须要检查Job的内存需求是否知足JobTracker的设置。

上面流程已经完毕，能够总结为下图：

TaskTracker

TaskTracker会周期性地经过心跳机制将本节点上资源的使用状况和任务的运行进度汇报给JobTracker，同时接收JobTracker发送过来的命令并执行相应的操做（如启动新任务、杀死 任务等）。TaskTracker 使用“slot”等量划分本节点上的资源量。 “slot”表明计算资源（CPU、 内存等）。一个 Task 获取到一个slot 后才有机会运行，而Hadoop调度器的做用就是将各个TaskTracker上的空闲slot分配给Task使用。slot分为Map slot和Reduce slot 两种，分别供Map Task和Reduce Task使用。TaskTracker经过slot数目（可配置参数）限定Task的并发度。

这里可能有人会混淆JobTracker、TaskTracker和Hadoop学习（一）——hdfs架构中所讲的的DataNode、NameNode。其实JobTracker对应于NameNode，TaskTracker对应于DataNode。DataNode和NameNode是针对数据存放来而言的，JobTracker和TaskTracker是对于MapReduce执行而言的。

MapTask和ReduceTask

Task 分为 Map Task 和 Reduce Task 两种，均由TaskTracker启动。从Hadoop学习（一）——hdfs架构中咱们知道，HDFS以固定大小的block 为基本单位存储数据，而对于MapReduce的输入而言，其处理单位是split。 split 与 block 的对应关系默认是1:1。split 是一个逻辑概念，它只包含一些元数据信息，好比数据起始位置、数据长度、数据所在节点等。它的划分方法彻底由用户本身决定。但须要注意的是，split的多少决定了Map Task的数目，由于每一个split会交由一个Map Task处理。split会在后面MapReduce的执行过程当中详细讲。

MapReduce 1.x工做流程

www.aboutyun.com/thread-1549…

整个MapReduce做业的工做过程，以下所示：

1. 做业的提交 JobClient的submitJob()方法实现的做业提交过程，以下所示： 经过JobTracker的getNewJobId()请求一个新的做业ID；（2） 检查做业的输出说明（好比没有指定输出目录或输出目录已经存在，就抛出异常）； 计算做业的输入分片（当分片没法计算时，好比输入路径不存在等缘由，就抛出异常）； 将运行做业所需的资源（好比做业Jar文件，配置文件，计算所得的输入分片等）复制到一个以做业ID命名的目录中。（集群中有多个副本可供TaskTracker访问）（3） 经过调用JobTracker的submitJob()方法告知做业准备执行。（4）

2. 做业的初始化 JobTracker接收到对其submitJob()方法的调用后，就会把这个调用放入一个内部队列中，交由做业调度器（好比先进先出调度器，容量调度器，公平调度器等）进行调度；（5） 初始化主要是建立一个表示正在运行做业的对象——封装任务和记录信息，以便跟踪任务的状态和进程；（5） 为了建立任务运行列表，做业调度器首先从HDFS中获取JobClient已计算好的输入分片信息（6）。而后为每一个分片建立一个MapTask，而且建立ReduceTask。（Task在此时被指定ID，请区分清楚Job的ID和Task的ID）。

3. 任务的分配 TaskTracker按期经过“心跳”与JobTracker进行通讯，主要是告知JobTracker自身是否还存活，以及是否已经准备好运行新的任务等；（7） JobTracker在为TaskTracker选择任务以前，必须先经过做业调度器选定任务所在的做业； 对于MapTask和ReduceTask，TaskTracker有固定数量的任务槽（准确数量由TaskTracker核的数量和内存大小来决定）。JobTracker会先将TaskTracker的MapTask填满，而后分配ReduceTask到TaskTracker； 对于MapTrask，JobTracker经过会选取一个距离其输入分片文件最近的TaskTracker。对于ReduceTask，由于没法考虑数据的本地化，因此也没有什么标准来选择哪一个TaskTracker。

4. 任务的执行 TaskTracker分配到一个任务后，经过从HDFS把做业的Jar文件复制到TaskTracker所在的文件系统（Jar本地化用来启动JVM），同时TaskTracker将应用程序所须要的所有文件从分布式缓存复制到本地磁盘；（8） TaskTracker为任务新建一个本地工做目录，并把Jar文件中的内容解压到这个文件夹中； TaskTracker启动一个新的JVM（9）来运行每一个Task（包括MapTask和ReduceTask），这样Client的MapReduce就不会影响TaskTracker守护进程（好比，致使崩溃或挂起等）； 子进程经过umbilical接口与父进程进行通讯，Task的子进程每隔几秒便告知父进程它的进度，直到任务完成。

5. 进程和状态的更新 一个做业和它的每一个任务都有一个状态信息，包括做业或任务的运行状态，Map和Reduce的进度，计数器值，状态消息或描述（能够由用户代码来设置）。这些状态信息在做业期间不断改变，它们是如何与Client通讯的呢？

• 任务在运行时，对其进度（即任务完成的百分比）保持追踪。对于MapTask，任务进度是已处理输入所占的比例。对于ReduceTask，状况稍微有点复杂，但系统仍然会估计已处理Reduce输入的比例；
• 这些消息经过必定的时间间隔由Child JVM—>TaskTracker—>JobTracker汇聚。JobTracker将产生一个代表全部运行做业及其任务状态的全局视图。能够经过Web UI查看。同时JobClient经过每秒查询JobTracker来得到最新状态，而且输出到控制台上。

6. 做业的完成 当JobTracker收到做业最后一个任务已完成的通知后，便把做业的状态设置为"成功"。而后，在JobClient查询状态时，便知道做业已成功完成，因而JobClient打印一条消息告知用户，最后从runJob()方法返回。

MapReduce 1.x的缺点

随着分布式系统集群的规模和其工做负荷的增加，原框架的问题逐渐浮出水面，主要的问题集中以下：

1. JobTracker 是 Map-reduce 的集中处理点，存在单点故障。
2. JobTracker 完成了太多的任务，形成了过多的资源消耗，当 map-reduce job 很是多的时候，会形成很大的内存开销，潜在来讲，也增长了 JobTracker fail 的风险，这也是业界广泛总结出老 Hadoop 的 Map-Reduce 只能支持 4000 节点主机的上限。
3. 在 TaskTracker 端，以 map/reduce task 的数目做为资源的表示过于简单，没有考虑到 cpu/ 内存的占用状况，若是两个大内存消耗的 task 被调度到了一块，很容易出现 OOM。
4. 在 TaskTracker 端，把资源强制划分为 map task slot 和 reduce task slot, 若是当系统中只有 map task 或者只有 reduce task 的时候，会形成资源的浪。
5. 源代码层面分析的时候，会发现代码很是的难读，经常由于一个 class 作了太多的事情，代码量达 3000 多行，，形成 class 的任务不清晰，增长 bug 修复和版本维护的难度。
6. 从操做的角度来看，如今的 Hadoop MapReduce 框架在有任何重要的或者不重要的变化 ( 例如 bug 修复，性能提高和特性化 ) 时，都会强制进行系统级别的升级更新。更糟的是，它无论用户的喜爱，强制让分布式集群系统的每个用户端同时更新。这些更新会让用户为了验证他们以前的应用程序是否是适用新的 Hadoop 版本而浪费大量时间。

MapReduce 2.x/Yarn框架

从业界使用分布式系统的变化趋势和 hadoop 框架的长远发展来看，MapReduce 的 JobTracker/TaskTracker 机制须要大规模的调整来修复它在可扩展性，内存消耗，线程模型，可靠性和性能上的缺陷。在过去的几年中，hadoop 开发团队作了一些 bug 的修复，可是最近这些修复的成本愈来愈高，这代表对原框架作出改变的难度愈来愈大。

YARN是Hadoop 2.0中的资源管理系统，它的基本设计思想是将MRv1中的JobTracker拆分红了两个独立的服务：一个全局的资源管理器ResourceManager和每一个应用程序特有的ApplicationMaster。其中ResourceManager负责整个系统的资源管理和分配，而ApplicationMaster负责单个应用程序的管理。ApplicationMaster 承担了之前的 TaskTracker 的一些角色，ResourceManager 承担了 JobTracker 的角色。

YARN是一个资源管理、任务调度的框架，主要包含三大模块：ResourceManager（RM）、NodeManager（NM）、ApplicationMaster（AM）。其中，ResourceManager负责全部资源的监控、分配和管理；ApplicationMaster负责每个具体应用程序的调度和协调；NodeManager负责每个节点的维护。对于全部的applications，RM拥有绝对的控制权和对资源的分配权。而每一个AM则会和RM协商资源，同时和NodeManager通讯来执行和监控task。 几个模块之间的关系如图所示。

MapReduce 2.x重点概念

ResourceManager(RM)

RM是一个全局的资源管理器，负责整个系统的资源管理和分配。它主要由两个组件构成：调度器（Scheduler）和应用程序管理器（Applications Manager，AsM）。

• 调度器

  调度器根据容量、队列等限制条件（如每一个队列分配必定的资源，最多执行必定数量的做业等），将系统中的资源分配给各个正在运行的应用程序。

  须要注意的是，该调度器是一个“纯调度器”，它再也不从事任何与具体应用程序相关的工做，好比不负责监控或者跟踪应用的执行状态等，也不负责从新启动因应用执行失败或者硬件故障而产生的失败任务，这些均交由应用程序相关的ApplicationMaster完成。调度器仅根据各个应用程序的资源需求进行资源分配，而资源分配单位用一个抽象概念“资源容器”（Resource Container，简称Container）表示，Container是一个动态资源分配单位，它将内存、CPU、磁盘、网络等资源封装在一块儿，从而限定每一个任务使用的资源量。此外，该调度器是一个可插拔的组件，用户可根据本身的须要设计新的调度器，YARN提供了多种直接可用的调度器，好比Fair Scheduler和Capacity Scheduler等。

• 应用程序管理器

  应用程序管理器负责管理整个系统中全部应用程序，包括应用程序提交、与调度器协商资源以启动ApplicationMaster、监控ApplicationMaster运行状态并在失败时从新启动它等。

NodeManager（NM）

NM是每一个节点上的资源和任务管理器，一方面，它会定时地向RM汇报本节点上的资源使用状况和各个Container的运行状态；另外一方面，它接收并处理来自AM的Container启动/中止等各类请求。

ApplicationMaster（AM）

用户提交的应用程序均包含一个AM，负责为应用程序申请资源并分配给内部的任务，应用的监控，跟踪应用执行状态，重启失败任务等。ApplicationMaster是应用框架，它负责向ResourceManager协调资源，而且与NodeManager协同工做完成Task的执行和监控。MapReduce就是原生支持的一种框架，能够在YARN上运行Mapreduce做业。有不少分布式应用都开发了对应的应用程序框架，用于在YARN上运行任务，例如Spark，Storm等。若是须要，咱们也能够本身写一个符合规范的YARN application。

Container

Container 是 YARN 中的资源抽象，它封装了某个节点上的多维度资源，如内存、CPU、磁盘、网络等，当AM向RM申请资源时，RM为AM返回的资源即是用Container表示的。YARN会为每一个任务分配一个Container，且该任务只能使用该Container中描述的资源。每一个Container能够根据须要运行ApplicationMaster、Map、Reduce或者任意的程序。

YARN应用工做流程

1. 用户向YARN中提交应用程序，其中包括AM程序、启动AM的命令、命令参数、用户程序等；事实上，须要准确描述运行ApplicationMaster的unix进程的全部信息。提交工做一般由YarnClient来完成。

2. RM为该应用程序分配第一个Container，并与对应的NM通讯，要求它在这个Container中启动AM；

3. AM首先向RM注册，这样用户能够直接经过RM査看应用程序的运行状态，运行状态经过 AMRMClientAsync.CallbackHandler的getProgress() 方法来传递给RM。 而后它将为各个任务申请资源，并监控它的运行状态，直到运行结束，即重复步骤4〜7；

4. AM采用轮询的方式经过RPC协议向RM申请和领取资源；资源的协调经过 AMRMClientAsync异步完成,相应的处理方法封装在AMRMClientAsync.CallbackHandler中。

5. —旦AM申请到资源后，便与对应的NM通讯，要求它启动任务；一般须要指定一个ContainerLaunchContext，提供Container启动时须要的信息。

6. NM为任务设置好运行环境(包括环境变量、JAR包、二进制程序等)后，将任务启动命令写到一个脚本中，并经过运行该脚本启动任务；

7. 各个任务经过某个RPC协议向AM汇报本身的状态和进度，以让AM随时掌握各个任务的运行状态，从而能够在任务失败时从新启动任务；ApplicationMaster与NM的通讯经过NMClientAsync object来完成，容器的全部事件经过NMClientAsync.CallbackHandler来处理。例如启动、状态更新、中止等。

8. 应用程序运行完成后，AM向RM注销并关闭本身。