Hadoop系统架构

一、绪论21 世纪的第一个十年里， 互联网高速发展， Web 的易用性、 实用性使它成为最为广泛、最有前途、最有魅力的信息传播技术。作为信息交互的载体，Web的特性催生了多种新兴产业，电子商务、 社交网络在最近几年发展尤为迅速。互联网用户也在过去的数年间增长迅速，根据我国互联网络信息中心发布的第27 次中国互联网络发展状况统计报告显示，截至2010 年 12 月底，我国网民规模达到 4.57 亿，较 2009 年底增加 7330 万人。报告中还显示，网络购物用户年增长 48.6%，是用户增长最快的应用。 2011 年以微博为代表的新型社交网站迅速成长，新浪微博在 2011 年 9 月注册用户达到 2.75 亿，微博用户平均每天发布的微博数达到 8600 万条。无论是电子商务网站中的产品图片展示，还是社交网站中的图片分享，其图片数量都在呈现几何基础的增长。以国内外几大 IT 巨头为例，截止至2011 年 6 月，Facebook用户已经上传了150 亿张照片，加上缩略图，总容量超过 L5PB。此外，每周新增照片为2.2 亿张，约 25TB。高峰期，Facebook每秒处理 55 万张照片 ! 国外最大的图片分享网站 Flickr共存储 4.7 亿张图片，而且相当多的图片是高清数码图片，单张图片大小4?5M左右，消耗 2PB 存储空间，每秒需要处理38000 次请求，每天新增图片超过40万。Flickr 采用的 squid 缓存了总计 3500 万张图片，内存中存储有 200 万张图片。淘宝网作为我国最大的电子商务平台， 在线商品达到 10 亿，图片服务器存储 286 亿张图片，总容量达到 1PB，且每天仍在以千万级别增长。由于图片表达信息远胜于文字描述， 所以电子商务尤其注重图片的显示质量、 上传时间、访问速度等问题。根据淘宝网的流量分析，整个淘宝网流量中，图片的访问流量达到90%以上。腾讯的相册用户总上传图片数 600 亿存储量 12PB、每周上传图片数 10 亿、存储 3 种规格 1300 亿图片，峰值访问每秒 50 万次。由于图片量非常大， 海量图片需要消耗海量的存储空间， 图片的存储和检索都会出现一定的瓶颈， 存储系统的快速访问、扩容性、容错性都将是存储系统设计的目标。由此可见，面对海量的图片，如何高效的存储、管理这些图片已经成为一个迫切需要解决的问题。NetApp，美国网域存储技术有限公司，是 IT 存储业界的佼佼者，倡导向数据密集型的企业提供统一的存储解决方案，用以整合在网络上来自服务器的数据，并有效管理呈爆炸性增长的数据。大多数IT 公司在面临海量数据存储问题的时候都会选择NetApp 公司提供的商用存储系统，淘宝网2007 前一直使用应用该公司的文件存储系统。 但随着图片文件数量以每年 2 倍的速度增长， NetApp 公司最高端的产品也不能满足淘宝网存储的要求。 商业存储服务的不足有以下几点：首先是文件数量太大，网络存储设备无法支撑，连接存储系统的服务器越来越多，网络连接数已经达到了网络存储设备的极限。其次是商用存储系统不能根据企业特定的业务进行存储和读取优化， 导致面对大规模的小文件存储与读取， 磁盘磁头需要频繁的寻道和换道， 造成读取上的延迟。再加上高并发访问量的背景，系统很容易出现问题。最后是花费问题，商业存储系统扩容成本太高， 10T 的存储容量需要几百万人民币。在面临海量存储需求的时候，高成本并没有带来高效率，高可靠性，高容错性。况且，过分依赖商业系统在维护性、创造性上受到商业公司约束，难以满足互联网企业的飞速发展。云计算的出现带动了技术发展朝向一个新的方向。 它创造性的根据分布式处理、并行处理和网格计算的发展， 提出了新的分布式集群技术。 云存储概是在云计算概念上延伸和发展出来的一个新的概念， 是指通过集群应用、 网格技术或分布式文件系统等功能， 将网络中大量各种不同类型的存储设备通过应用软件集合起来协同工作， 共同对外提供数据存储和业务访问功能的一个系统。 当云计算系统运算和处理的核心是大量数据的存储和管理时， 云计算系统中就需要配置大量的存储设备，那么云计算系统就转变成为一个云存储系统， 所以云存储是一个以数据存储和管理为核心的云计算系统。云存储的概念改变了存储领域， 可以尝试以相对廉价的存储设备部署在云端作为存储服务器，利用分布式文件系统统一管理， 对外提供存储和业务访问接口。由云端提供存储服务， 达到业务与存储的解稱合， 不仅能根据不同业务背景设定不同的存储、访问接口，优化存取服务， 还能将容灾和安全性固定在云端， 此外，由于采用分布式文件系统，云端服务器扩展相对容易。二、 Hadoop 云计算系统Hadoop 是一个分布式系统基础架构，由Apache 基金会开发。作为Google一系列产品的幵源实现，是一个更容易开发和运行处理大规模数据的软件平台。 Hadoop 中包含一系列相关的子项目， 这些项目都隶属于 Apache 软件基金会。最著名的是并行计算模型 （ MapReduce）和分布式文件系统 (HDFS) ，其他的子系统提供了一些附加功能，或者在 core 上增加了一些高级的抽象。Core：分布式系统和通用 I/O 组件和接口，支持序列化、 Java 远程过程调用等等。Avro：支持跨语言过程调用，持久数据存储的数据序列化系统。MapReduce：构建在廉价的PC 机器上的分布式数据处理模型和运行环境。HDFS：构建在廉价的PC 机器上的分布式文件系统。Pig：处理海量数据集的数据流语言和运行环境。pig 运行在HDFS 和MapReduce之上。HBase：面向列的分布式数据库。 HBase 使用 HDFS 作为底层存储，同时使用 MapReduce支持批处理模式的计算和随机查询。ZooKeeper：提供分布式、高效的协作服务。 ZooKeeper 提供分布式锁这样的原子操作，可以用来构建分布式应用。Hive：分布式数据仓库， Hive 使用 HDFS 存储数据，提供类似 SQL 的语言 (转换为 MapReduce任务 )查询数据。Chukwa：分布式数据采集和分析系统。 使用 HDFS 存储数据，使用 Mapreduce 输出分析报告。三、分布式文件系统（HDFS）Hadoop 分布式文件系统 HDFS 被设计成稳定的存储海量数据，并且适合运行在普通硬件上的分布式文件系统。 HDFS 能提供高吞吐量的数据访问性能，给用户的应用提供更高的带宽， 非常适合海量数据集上的应用。 它运行于廉价的普通硬件之上， 但是可以提供可靠性、 稳定的容错功能。 面对海量数据和大量用户仍然可以提供高质量的服务，具有优秀的并发处理的能力。3.1 HDFS 的特点(1) HDFS 认为硬件错误是一种正常的现象。 HDFS 由成百上千的普通硬件构成的服务器所组成，其中每个服务器上都存储着文件系维的数据块。 HDFS 文件系统非常庞大， 其中的任何一个服务器都可能出现故障， 这些服务器就会处于故障状态，从而导致系统的一部分数据丢失， 而且有一部分机器损坏之后可能无法恢复到正常工作状态。所以及时的检查、错误检测、数据备份容错、自动恢复是HDFS 分布式文件系统非常重要的功能。 HDFS 分布式文件系统通过自己的检测协议定时自动监控系统全部机器的状态， 一旦检测到机器故障等问题， 能够迅速地检测，定位、冗余并恢复这些故障机器上的数据。 基于以上设计的 HDFS 就具有错误检测和快速、自动恢复数据的功能。(2) 在 HDFS 上运行的应用需要以流式访问它们的数据集。HDFS 具有很好的数据批处理能力。 HDFS 更注重用于数据访问的高吞吐量，而对于数据访问的延迟和响应时间要求不做很严格处理。(3) HDFS 上的应用一般都是处理海量数据集的程序。 HDFS 上的文件大小一般都在 GB 至 TB 的大小。 HDFS 可以非常好的支持大文件存储。通过利用分布式集群 HDFS 能提供非常高的数据传输带宽， HDFS 集群可以扩展到数百个节点。同时一个 HDFS 文件系统可以支撑数以千万计的文件。 HDFS 分布式文件系统可以处理快速增长的、包含数以万计的对象、长度达 TB 的数据集，也可以管理成千上万的 KB 规模的文件块。(4) HDFS 采用一次写入多次读取的方式。在 HDFS 系统中一个文件经过创建、写入和关闭之后就不允许再去修改这个文件， 简化了数据一致性问题， 实现了高吞吐量访问数据的能力。 一般情况下，每次写入的数据的大小和大规模读取的模型基本一样， 数据一旦被写入后， 文件就不允许被修改了。 同时系统也支持小规模的随机位置写入操作。 MapReduce应用和网络爬虫应用是适应这个模型的最好应用说明。(5) 通常应用请求的计算的数据附近化是最高效的，处理海量数据的时候做到计算和数据距离最近可以得到最高的处理效率。 所以 HDFS 具有计算程序优先选择距离最近的数据的策略。 如果遇到网络阻塞将对计算程序访问数据的速度产生影响，采用附近化策略可以避免这种情况， 同时可以提高系统整体处理数据的吞吐量。把计算程序放到数据附近比把数据移动到计算的附近更高效。 HDFS 为提供了把应用程序移动到数据附近的接口。(6) HDFS 具有非常好的平台可移植性。 HDFS 使用 JAVA 开发， JAVA 本身就具有跨平台的特性。 HDFS 的可移植性推动它在大规模数据应用领域上的应用。同时 HDFS 提供其他语言的接口，方便用户使用。HDFS 分布式文件系统的以上特点可以充分保证数据的可靠性、安全性，保证系统的多并发和高速处理海量数据的能力，同时基于以上的策略，HDFS 分布式文件系统可以保证数据的一致性和自动修复，保证海量数据的安全和具有很好的存储性能。3.2 HDFS 系统架构HDFS 采用 Master/Slave 的主从结构。一个HDFS 集群是由一个主控节点（ Namenode）和一定数量的数据节点（ Datanode）组成的，如图 1 所示。主控节点是一个中心服务器， 是整个文件系统的大脑， 它负责管理文件系统的命名空间（ Namespace）和客户端对文件的访问。数据节点在集群中一般是一个节点对应一台服务器， 负责管理节点上它们所附带的存储。 在内部，一个文件其实分成一个或多个数据块， 这些块存储在数据节点集合中。 主控节点执行文件系统的命名空间操作，例如打开、关闭、重命名文件和目录，同时决定数据块到具体数据节点的映射。数据节点在主控节点的指挥下进行块的创建、删除和复制。主控节点和数据节点都是被设计成可以运行在普通的廉价的运行Linux 的机器上。 HDFS 采用 Java 语言开发，因此可以部署在很大范围的机器上。一个典型的部署场景是一台机器跑一个单独的主控节点， 集群中的其他机器各跑一个数据节点实例。单一主控节点大大简化了系统的架构。主控节点负责管理所有的 HDFS 元数据，客户端传输文件数据时就不需要通过主控节点，而是直接与数据节点建立连接。图 1 HDFS 系统架构的数据分布一个文件系统中，最重要的数据，其实就是整个文件系统的目录结构和具体每个文件的数据。 具体的文件数据被切分成数据块，存放在数据服务器上。 每一个文件数据块，在数据服务器上都表征为一对文件（普通的 Linux 文件），一个是数据文件，一个是附加信息的元数据文件， 我们把这对文件简称为数据块文件。数据块文件存放在数据目录下，它有一个名为current 的根目录，然后里面有若干个数据块文件和从 dir0-dir63 的最多 64 个的子目录，子目录内部结构等同于 current 目录，依次类推。相比数据服务器，主控服务器的数据量不大，但逻辑非常复杂。主控服务器主要有三类数据： 文件系统的目录结构数据， 各个文件的分块信息， 数据块的位置信息（就数据块放置在哪些数据服务器上） 。在 HDFS 架构中，只有文件的目录结构和分块信息才会被持久化到本地磁盘上， 而数据块的位置信息则是通过动态汇总过来的，仅仅存活在内存数据结构中，机器挂了，就灰飞烟灭了。每一个数据服务器启动后， 都会向主控服务器发送注册消息， 将其上数据块的状况都告知于主控服务器。的数据组织兼容 HDFS 的应用都是处理大数据集合的。这些应用都是写数据一次，读却是一次到多次，并且读的速度要满足流式读。 HDFS 支持文件的一次写入多次读取（ write-once-read-many）语义。一个典型的数据块（ block）大小是 64MB ，因而，文件总是按照64M 切分成 chunk，每个 chunk 存储于不同的数据节点。某个客户端创建文件的请求其实并没有立即发给主控节点，事实上，HDFS客户端会将文件数据缓存到本地的一个临时文件。应用的写被透明地重定向到这个临时文件。当这个临时文件累积的数据超过一个block 的大小（默认64M) ，客户端才会联系主控节点。 主控节点将文件名插入文件系统的层次结构中，并且分配一个数据块给它， 然后返回数据节点的标识符和目标数据块给客户端。 客户端将本地临时文件 flush 到指定的数据节点上。当文件关闭时，在临时文件中剩余的没有 flush 的数据也会传输到指定的数据节点，然后客户端告诉主控节点文件已经关闭。此时主控节点才将文件创建操作提交到持久存储。 如果主控节点在文件关闭前挂了，该文件将丢失。的数据复制HDFS 被设计成在一个大集群中可以跨机器地可靠地存储海量的文件。它将每个文件存储成block 序列，除了最后一个block，所有的 block 都是同样的大小。文件的所有block 为了容错都会被复制。每个文件的block 大小和复制因子（ replication）都是可配置的。复制因子可以在文件创建的时候配置，以后也可以改变。 HDFS 中的文件是一次写入（ write-one），并且严格要求在任何时候只有一个 writer 。主控节点全权管理 block 的复制，它周期性地从集群中的每个数据节点接收心跳包和一个数据块报告（ Blockreport）。心跳包的接收表示该数据节点正常工作，而块报告包括了该数据节点上所有数据块组成的列表。图 2 HDFS 的数据复制的通信协议所有的 HDFS 通讯协议都是构建在 TCP/IP 协议上。客户端通过一个可配置的端口连接到主控节点，通过客户端协议（ ClientProtocol ）与主控节点交互。而数据节点是使用数据节点协议（ DatanodeProtocol）与 主控 节点 交互 。 从ClientProtocol 和 Datanodeprotocol 抽象出一个远程调用 (RPC），在设计上，主控节点不会主动发起 RPC，而是响应来自客户端和数据节点的 RPC 请求。如图 3 所示：图 3 HDFS 的通信协议3.3 HDFS 的初始化与数据存取过程的启动过程HDFS 集群中，一般只有主控节点和数据节点两种节点，所以HDFS 的启动就是数据节点和主控节点的启动。HDFS 启动过程：首先是主控节点最先启动，主控节点必须在所有数据节点之前启动， 而且在主控节点启动之后， 数据节点也不会马上就启动， 数据节点需要在主控节点完成必要的操作之后才开始启动。 主控节点和数据节点的启动过程如图 4 所示。图 4 HDFS 启动过程主控节点启动时会先创建 Server，Server 是 RFC 服务器端的实现， 主要负责和客户端通信， 并对远程调用中的参数和返回值进行反序列化和序列化。 主控节点真正负责执行远程方法。 FSNamesystem保存了全部的关于文件系统的元数据信息和操作日志， 操作日志负责数据的持久化来保证系统的可恢复性。 然后再创建 FSNamesystem，主控节点在创建 FSNamesystem的同时会把元数据信息加载到内存里，由于加载元数据到内存非常耗费时间， 所以主控节点启动之后， 其它的数据节点不能马上启动， 需要等到主控节点加载完元数据之后的某个时机开始启动。加载完成后，主控节点开启 Server 的远程调用服务。在这之后主控节点会进入安全模式， 安全模式下主控节点不接受任何数据的写入和读取， 即不为客户端提供任何服务。 然后主控节点开始等待数据节点的注册和通信， 数据节点此时上报数据块的 Block 信息以及自己本身的一些状态信息， 这些信息为以后的存储策略服务。当在一定的时间间隔之后没有收到数据节点新的注册， 主控节点就认为集群中没有其它没有注册的数据节点了， 主控节点就会离开安全模式进入正常模式为客户端服务。在主控节点启动的过程中主要是自身的初始化和数据节点的注册，此时数据节点需要向主控节点汇报自身的一些状态信息，这些信息为数据存储策略服务。数据节点的启动的时间必须要在主控节点启动之后，主控节点RPC 的 Server 服务开启之后，数据节点才能开始启动。数据节点主要有DataStorage、Server、DataXceiverServer、 DataNodeProtocol 四个服务组件，其中 Datastorage保存数据块信息， DataNodeProtocol负责调用主控节点的服务， DataXceiverServer 负责客户端和数据节点之间的数据传输，数据节点的Server 负责为客户端和其它数据节点提供服务。HDFS 集群启动时，每个数据节点都要向主控节点发送注册的请求，在请求通过后才可以加入HDFS 集群中。数据节点调用 DatanodeProtocol协议向主控节点进行注册，数据节点向主控节点注册有两个目的： 首先是通告主控节点其提供的服务的网络地址端口， 其次是获取数据节点对其的管理与控制。 每一个客户端无需获取集群中所有的数据节点的服务地址， 主控节点会记录所有的数据节点信息。客户端通过主控节点来获取需要访问的数据节点的信息即可。 主控节点记录所有的数据节点汇报的状态信息， 根据数据节点的状态信息调整集群的负载均衡与任务的合理安排和调度。在 HDFS 启动过程中，数据节点需要向控制节点汇报状态信息，这些信息包括磁盘容量， 块状态等信息。 数据节点的状态信息需要定时和控制节点汇报。 这些信息是通过一种叫做心跳协议的方式从数据节点汇报给控制节点的， 心跳协议能够及时的汇报每个数据节点的状态信息， 通过设计优化的心跳协议可以向控制节点汇报更多的状态信息。 通过汇报的状态信息， 可以为节点选择策略提供更多依据，基于更多的依据就可以更加准确的判断一个节点的负载， 可以保证选择的节点是比较空闲的数据节点。的数据存取过程客户端访问 HDFS 一般都是通过调用Hadoop 提供的 API 实现的，而底层数据操作的过程对客户端是透明的。下面分别从数据读取和写入两方面介绍对 HDFS 数据操作进行剖析。(1) HDFS 文件读取剖析客户端读取 HDFS 中数据的流程如图5 所示：图 5 HDFS 数据读取流程图客户端通过调用FileSystem 对象的 open()方法来打开希望读取的文件（步骤1），对于 HDFS 来说，这个对象是分布式文件系统的一个实例。 DistributedFileSystem 对象通过 RPC 来调用控制节点，以确定文件起始块的位置（步骤 2）。对于每一个块，控制节点返回存有该块复本的数据节点地址。此外，这些数据节点根据它们与客户端的距离来排序（根据集群的网络拓扑） 。如果该客户端本身就是一个数据节点（比如在一个 MapReduce任务中），并保存有相应数据块的一个复本时，该节点将从本地数据节点中读取数据。DistributedFileSystem 对象返回一个 FSDataInputStream对象（一个支持文件定 位的 输 入 流 ）给 客户 端并 读 取数 据 。 FSDataInputStream 类 中 封 装 了 DFSInputStream 对象，该对象管理着数据节点和控制节点的 I/O 操作。接着，客户端对这个输入流调用read()方法（步骤 3）。存储着文件起始块的数据节点地址的 DFSInputStream 随即连接距离最近的数据节点。通过对数据流反复调用 read()方法，即可将数据从数据节点传输到客户端（步骤 4），达到块的末端时，DFSInputStream会关闭与该数据节点的连接， 然后寻找下一个块的最佳数据节点（步骤 5）。客户端只需要读取连续的流，上面这些对于客户端都是透明的。客户端从流中读取数据时， 块是按照打开 DFSInputStream 与数据节点新建连接的顺序读取的。它也需要询问控制节点来检索下一批所需块的数据节点的位置。一旦客户端完成读取，就对FSDataInputStream调用 close()方法（步骤 6）。在读取数据的时候， 如果 DFSInputStream在与数据节点通信时遇到错误， 它便会尝试从这个块的另外一个最邻近的数据节点读取数据。 它也会记住那个出现故障的数据节点，以保证以后不会反复读取该节点上后续的块。 DFSInputStream 也会通过校验和确认从数据节点发来的数据是否完整。如果发现一个损坏的块，它就会在 DFSInputStream 试图从其他数据节点中读取一个块的复本之前通知控制节点。(2) HDFS 文件写入剖析客户端将数据写入到HDFS 的流程如图 6 所示：图 6 HDFS 数据写入流程图客户端通过 DistributedFileSystem 对象调用 create()方法来创建文件（步骤 1）， DistributedFileSystem 对控制节点创建一个 RPC 调用，在文件系统的命名空间中创建一个新文件，此时文件中还没有相应的数据块（步骤 2）。控制节点执行各种不同的检查以确保这个文件不存在， 并且客户端有创建该文件的权限。 如果这些检查均通过， 控制节点就会为创建新文件增加一条记录； 否则文件创建失败并向客户端抛出一个 IOException 异常。 DistributedFileSystem 向客户端返回一个FSDataOutputStream对象，客户端通过此对象便可以写入数据。 与读取数据类似，FSDataOutputStream中封装着一个 DFSOutputStream对象，该对象负责处理与数据节点和控制节点之间的通信。在客户端写入数据时（步骤3），DFSOutputStream将需要写入的数据分成一个个的数据包，并写入内部队列（也称为数据队列- data queue）。DataStreamer负责处理数据队列， 他的责任是根据数据节点列表来要求控制节点分配合适的新块存储数据备份。这一组数据节点构成一个管线（假设复本数为3），所以管线中有 3 个节点。 DataStreamer将数据包流式传输到管线中的第1 个数据节点，该数据节点存储收到的数据包并将它发送到管线中的第2 个数据节点。同样第2个数据节点也存储该数据包并且发送给管线中的第3 个数据节点（步骤4）。DFSOutputStream 也维护着一个内部数据包队列来等待数据节点的收到确认回执，称为“确认队列” (ack queue)。当收到管线中所有数据节点的确认信息后，该数据包才会从确认队列中移除（步骤 5）。如果在数据写入期间数据节点发生故障，则执行以下操作，这对于写入数据的客户端是透明的。 首先关闭管线， 确认把队列中的任何数据包都添加回数据队列的最前端，以确保故障节点下游的数据节点不会漏掉任何一个数据包。 为存储在另一个正常数据节点的当前数据块指定一个新的标识， 并将该标识传送给控制节点，以便故障数据节点在恢复后可以删除存储的部分数据块。 从管线中删除故障数据节点并且把余下的数据块写入到管线中两个正常的数据节点。 控制节点注意到块复本量不足时， 会在另一个节点上创建一个新的复本。 后续的数据块继续正常接收处理。客户端完成数据的写入后， 会对数据流调用 close()方法（步骤 6），该操作将剩余的所有数据包写入到数据节点管线中， 并向控制节点发送文件写入完成的信号消息，等待确认。控制节点已经知道文件由那些块组成（通过 DataStreamer 获取数据块的分配信息） ，所以它在返回成功之前只需要等待数据块进行最小量的复制。四、并行计算模型（MapReduce）4.1 MapReduce 系统架构与 HDFS 类似， Hadoop 的 MapReduce集群也由三类服务器构成。其中作业服务器，在 Hadoop 中称为 JobTracker。作业服务器负责管理运行在此框架下所有作业，同时它也是为各个作业分配任务的核心。与HDFS 的主控服务器NameNode 类似，它也是作为单点存在的，简化了负责同步的流程。具体负责执行用户定义操作的是任务服务器 TaskTracker，每一个作业被拆分成很多个任务，包括 Map 任务和 Reduce 任务等，任务是具体的执行单元，它们都需要分配到合适任务服务器上去执行， 任务服务器一边执行一边向作业服务器汇报各个任务的状态，以此来帮助作业服务器了解作业执行的整体情况，分配新的任务等。除了作业的管理者和执行者，还需要有一个任务的提交者，这就是客户端Client 。与分布式文件系统一样，客户端也不是一个单独的进程，而是一组 API ，用户需要自定义需要的内容， 经由客户端相关的代码， 将作业及其相关内容和配置提交到作业服务器上，并时刻监控作业的执行状况。同样作为 Hadoop 的实现，与 HDFS 的通信机制类似， Hadoop 的 MapReduce也使用了协议接口来实现服务器间的交流。实现者作为RPC 服务器，调用者经由 RPC 代理进行调用，通过这种方式完成大部分的通信。Hadoop 的 MapReduce计算模型系统架构如图 7 所示：图 7 MapReduce 系统架构4.2 MapReduce 计算流程整个 MapReduce 作业的计算流程应该是：作业的提交；Map 任务的分配和执行； Reduce 任务的分配和执行；作业的完成。而在每个任务的执行中，又包含 输入的准备、算法的执行、输出的生成 三个子步骤。沿着这个流程，我们可以很快的整理清晰整个 MapReduce 框架下作业的执行。作业的提交一个作业，在提交之前，需要把所有应该配置的东西都配置好，因为一旦提交到了作业服务器上， 就陷入了完全自动化的流程， 用户除了观望， 最多也就能起一个监督作用，惩治一些不好好工作的任务。基本上，用户在提交代码阶段，需要做的工作主要是这样的：首先，书写好所有自定的代码， 最起码，需要有 Map 和 Reduce的执行代码。在 Hadoop 中，Map 需要派生自 Mapper 接口， Reduce需要派生自 Reducer接口。这里都是用的泛型，用以支持不同的键值类型。这两个接口都仅有一个方法，一个是 map，一个是 reduce，这两个方法都直接受四个参数，前两个是输入的键和值相关的数据结构， 第三个是作为输出相关的数据结构，最后一个， 是一个 Reporter 类的实例，实现的时候可以利用它来统计一些计数。除了这两个接口，还有大量可以派生的接口，比如分割的Partitioner 接口。然后，需要书写好主函数的代码， 其中最主要的内容就是实例化一个 JobConf 类的对象，然后调用其丰富的 setXXX 接口，设定好所需的内容，包括输入输出的文件路径， Map 和 Reduce的类，甚至包括读取写入文件所需的格式支持类，等等。最后，调用 JobClient 的 runJob 方法，提交此 JobConf 对象。 runJob 方法会先行调用到 JobSubmissionProtocol接口所定义的 submitJob 方法，将此作业，提交给作业服务器。接着， runJob 开始循环，不停的调用 JobSubmissionProtocol的 getTaskCompletionEvents方法，获得 TaskCompletionEvent 类的对象实例，了解此作业各任务的执行状况。任务的分配和执行当一个作业，从客户端提交到了作业服务器上，作业服务器会生成一个JobInProgress对象，作为与之对应的标识， 用于管理。作业被拆分成若干个 Map 任务后，会预先挂在作业服务器上的任务服务器拓扑树。 这是依照分布式文件数据块的位置来划分的，比如一个 Map 任务需要用某个数据块，这个数据块有三份备份，那么，在这三台服务器上都会挂上此任务，可以视为是一个预分配。关于任务管理和分配的大部分的真实功能和逻辑的实现，JobInProgress 则依托JobInProgressListener和 TaskScheduler的子类。 TaskScheduler，顾名思义是用于任务分配的策略类。它会掌握好所有作业的任务信息，其 assignTasks函数，接受一个 TaskTrackerStatus作为参数，依照此任务服务器的状态和现有的任务状况，为其分配新的任务。而为了掌握所有作业相关任务的状况， TaskScheduler 会将若干个 JobInProgressListener注册到 JobTracker中去，当有新的作业到达、移除或更新的时候， JobTracker会告知给所有的 JobInProgressListener，以便它们做出相应的处理。任务分配是一个重要的环节，所谓任务分配，就是将合适作业的合适任务分配到合适的服务器上。不难看出，里面蕴含了两个步骤，先是选择作业，然后是在此作业中选择任务。 和所有分配工作一样， 任务分配也是一个复杂的活。 不良好的任务分配，可能会导致网络流量增加、某些任务服务器负载过重效率下降，等等。不仅如此，任务分配还是一个无一致模式的问题，不同的业务背景，可能需要不同的算法才能满足需求。根据具体的分配算法确定了从哪个作业提取任务后，经过一系列的调用，最后实际是由 JobInProgress的 findNewMapTask 函数完成的。 它的算法很简单， 就是尽全力为此服务器分配且尽可能好的分配任务， 也就是说，只要还有可分配的任务，就一定会分给它， 而不考虑后来者。 作业服务器会从离它最近的服务器开始，看上面是否还挂着未分配的任务（预分配上的） ，从近到远，如果所有的任务都分配了， 那么看有没有开启多次执行， 如果开启，考虑把未完成的任务再分配一次。对于作业服务器来说，把一个任务分配出去了，并不意味着它就彻底解放，可以对此任务可以不管不顾了。因为任务可能在任务服务器上执行失败，也可能执行缓慢，这都需要作业服务器帮助它们再来一次。因此在 Task 中，记录有一个 TaskAttemptID，对于任务服务器而言，它们每次跑的，其实都只是一个 Attempt而已， Reduce任务只需要采信一个的输出，其他都算白忙乎了。与 HDFS 类似，任务服务器是通过心跳消息，向作业服务器汇报此时此刻其上各个任务执行的状况，并向作业服务器申请新的任务的。具体实现，是TaskTracker调用 InterTrackerProtocol 协议的 heartbeat方法来做的。这个方法接受一个 TaskTrackerStatus对象作为参数， 它描述了此时此任务服务器的状态。 当其有余力接受新的任务的时候，它还会传入 acceptNewTasks为 true 的参数，表示希望作业服务器委以重任。 JobTracker接收到相关的参数后，经过处理，会返回一个 HeartbeatResponse对象。这个对象中，定义了一组 TaskTrackerAction，用于指导任务服务器进行下一步的工作。当 TaskTracker 收到的 TaskTrackerAction 中，包含了 LaunchTaskAction，它会 开 始 执 行 所 分 配 的 新 的 任 务 。 在 TaskTracker 中 ， 有 一 个TaskTracker.TaskLauncher线程（确切的说是两个，一个等Map 任务，一个等Reduce任务），它们在痴痴的守候着新任务的来到。一旦等到了，会最终调用到 Task 的 createRunner方法，构造出一个 TaskRunner对象，新建一个线程来执行。对于一个 Map 任务，它对应的 Runner 是 TaskRunner的子类 MapTaskRunner，不过，核心部分都在 TaskRunner 的实现内。 TaskRunner会先将所需的文件全部下载并拆包好， 并记录到一个全局缓存中， 这是一个全局的目录， 可以供所有此作业的所有任务使用。 它会用一些软链接， 将一些文件名链接到这个缓存中来。 然后，根据不同的参数，配置出一个 JVM 执行的环境，这个环境与 JvmEnv 类的对象对应。接着，TaskRunner会调用 JvmManager的 launchJvm 方法，提交给 JvmManager 处理。 JvmManager 用于管理该 TaskTracker上所有运行的 Task 子进程。在目前的实现中，尝试的是池化的方式。有若干个固定的槽，如果槽没有满，那么就启动新的子进程，否则，就寻找idle 的进程，如果是同 Job 的直接放进去，否则杀死这个进程，用一个新的进程代替。每一个进程都是由 JvmRunner 来管理的，它也是位于单独线程中的。任务的分配和执行比之 Map 任务， Reduce的分配及其简单，基本上是所有Map 任务完成了，有空闲的任务服务器，来了就给分配一个Job 任务。因为 Map 任务的结果星罗棋布，且变化多端，真要搞一个全局优化的算法，绝对是得不偿失。而Reduce任务的执行进程的构造和分配流程，与Map 基本完全的一致。但其实， Reduce任务与 Map 任务的最大不同，是Map 任务的文件都在本地隔着，而 Reduce任务需要到处采集。 这个流程是作业服务器经由此 Reduce任务所处的任务服务器，告诉 Reduce任务正在执行的进程，它需要的 Map 任务执行过的服务器地址，此 Reduce任务服务器会于原 Map 任务服务器联系（如果是本地就免了），通过 FTP 服务，下载过来。这个隐含的直接数据联系，就是执行Reduce任务与执行 Map 任务最大的不同了。作业的完成当所有 Reduce 任务都完成了，所需数据都写到了分布式文件系统上，整个作业就算正式完成了。图2-8 模拟了一个由 3 个 Map 任务和 1 个 Reduce任务构成的作业执行流程。 我们可以看到， 在执行的过程中， 只要有人太慢， 或者失败，就会增加一次尝试，以此换取最快的执行总时间。 一旦所有 Map 任务完成，Reduce开始运作（实际上，目前Hadoop 已做到一定数量的Map 任务完成就可以启动Reduce任务了），对于每一个 Map 任务来说，只有执行到Reduce任务把它上面的数据下载完成，才算成功，否则，都是失败，需要重新进行尝试。综上所述， Hadoop 的 MapReduce的整个计算流程可由图2-9 进行表示。图 9 MapReduce 并行计算流程