hadoop 讲解

Posted 2021-01-21 lzc-1105m

大数据的5V特点（IBM提出）：

Volume（大量）、Velocity（高速）、Variety（多样）、Value（低价值密度）、Veracity（真实性）
大数据的应用：

预测犯罪的发生；预测禽流感的散布；美国选举结果；利用手机定位数据和交通数据建立城市规划；电商把假货卖给谁等等。
大数据涉及到的技术：

数据采集；

数据存储；

数据处理/分析/挖掘；

可视化。
Hadoop

Hadoop是一个由Apache基金会所开发的分布式系统基础架构。

Hadoop实现了一个分布式文件系统（Hadoop Distributed File System），简称HDFS。HDFS有高容错性的特点，并且设计用来部署在低廉的（low-cost）硬件上；而且它提供高吞吐量（high throughput）来访问应用程序的数据，适合那些有着超大数据集（large data set）的应用程序。HDFS放宽了（relax）POSIX的要求，可以以流的形式访问（streaming access）文件系统中的数据。

Hadoop的框架最核心的设计就是：HDFS和MapReduce。HDFS为海量的数据提供了存储，则MapReduce为海量的数据提供了计算。
什么是分布式系统？

分布式系统是若干独立计算机的集合，这计算机对用户来说就像单个相关系统。也就是说分布式系统背后是由一系列的计算机组成的，但用户感知不到背后的逻辑，就像访问单个计算机一样。

一个标准的分布式系统应该具有以下几个主要特征：

    分布性

分布式系统中的多台计算机之间在空间位置上可以随意分布，系统中的多台计算机之间没有主、从之分，即没有控制整个系统的主机，也没有受控的从机。

    透明性

系统资源被所有计算机共享。每台计算机的用户不仅可以使用本机的资源，还可以使用本分布式系统中其他计算机的资源(包括CPU、文件、打印机等)。

    同一性

系统中的若干台计算机可以互相协作来完成一个共同的任务，或者说一个程序可以分布在几台计算机上并行地运行。

    通信性

系统中任意两台计算机都可以通过通信来交换信息。

在分布式系统中：

1、应用可以按业务类型拆分成多个应用，再按结构分成接口层、服务层；我们也可以按访问入口分，如移动端、PC端等定义不同的接口应用；

2、数据库可以按业务类型拆分成多个实例，还可以对单表进行分库分表；

3、增加分布式缓存、搜索、文件、消息队列、非关系型数据库等中间件；

很明显，分布式系统可以解决集中式不便扩展的弊端，我们可以很方便的在任何一个环节扩展应用，就算一个应用出现问题也不会影响到别的应用。

分布式系统虽好，也带来了系统的复杂性，如分布式事务、分布式锁、分布式session、数据一致性等都是现在分布式系统中需要解决的难题，虽然已经有很多成熟的方案，但都不完美。分布式系统也增加了开发测试运维成本，工作量增加，分布式系统管理不好反而会变成一种负担。

参考：https://blog.csdn.net/youanyyou/article/details/79406507
Hadoop能做什么？

建立大型仓库，PB级（1PB=1024TB 1TB=1024GB）数据的存储、处理、分析、统计等业务；

搜索引擎，日志分析，商业智能，数据挖掘等。

1，HDFS：

HDFS：分布式文件系统，是Hadoop体系中数据存储管理的基础。它是一个高度容错的系统，能检测和应对硬件故障，用于在低成本的通用硬件上运行。HDFS简化了文件的一致性模型，通过流式数据访问（1一次写入，多次读取。文件一旦写入不能修改，只能追加。2，它能保证数据的一致性。），提供高吞吐量应用程序数据访问功能，适合带有大型数据集的应用程序。

将文件切分成指定大小的数据块并以多副本（默认：128M）的存储在多个计算机上；

数据切分、多副本、容错等操作是对用户透明的。

特点：扩展性&容错性&海量数量存储

HDFS 缺点（不适用适用HDFS的场景）：
1、低延时数据访问
        1)比如毫秒级的来存储数据，这是不行的，它做不到。
       2)它适合高吞吐率的场景，就是在某一时间内写入大量的数据。但是它在低延时的情况下是不行的，比如毫秒级以内读取数据，这样它是很难做到的。

2、小文件存储
        1)存储大量小文件的话，它会占用 NameNode大量的内存来存储文件、目录和块信息。这样是不可取的，因为NameNode的内存总是有限的。
        2)小文件存储的寻道时间会超过读取时间，它违反了HDFS的设计目标。
3、并发写入、文件随机修改

        1)一个文件只能有一个写，不允许多个线程同时写。
        2)仅支持数据 append（追加），不支持文件的随机修改。

简单的分布式系统设计：

HDFS构架原则：

    元数据与数据分离：文件本身的属性（即元数据）与文件所持有的数据分离
    主/从架构：一个HDFS集群是由一个NameNode和一定数目的DataNode组成
    一次写入多次读取：HDFS中的文件在任何时间只能有一个Writer。当文件被创建，接着写入数据，最后，一旦文件被关闭，就不能再修改
    移动计算比移动数据更划算：数据运算，越靠近数据（为了尽量减小全局带宽的消耗和读延时），执行运算的性能就越好，由于hdfs数据分布在不同机器上，要让网络的消耗最低，并提高系统的吞吐量，最佳方式是将运算的执行移到离它要处理的数据更近的地方，而不是移动数据

数据复制

HDFS被设计为用于在大型集群中的服务器之间可靠地存储非常大的文件。它把每一个文件存储为序列的块，这些序列块会被复制用于容错，每个文件的块大小和复制因子是可以配置的。同一文件中除了最后一个块之外，其它块的大小都是相同的，而在DistributedFileSystem.append和HdfsDataOutputStream.hsync方法支持参数flag后，用户可以随时新建一个块而不是默认地往最后一个块添加数据直到配置的块大小上限为止。

应用程序可以指定文件的副本数，复制因子可以在创建文件的时候指定，也可以在创建之后修改。HDFS中的文件是一次写入的（除了追加和截断之外），并且在任何时候都只有一个写操作。NameNode负责块复制的所有决策，它周期地接收集群中每个DataNode的心跳和块报告Blockreport，收到心跳意味着DataNode是正常工作的。一份Blockreport包含该DataNode所有块的列表。
副本存放位置

副本的存放位置是会严重影响HDFS的可靠性和性能，它的优化是HDFS区分于其他分布式文件系统的的特点。HDFS使用机架感知rack-aware副本放置策略，它的意图是提高数据可靠性，可用性和网络带宽的利用率。目前的副本放置策略的实现是朝着这个方向努力的第一步，实现这一策略的短期目标是在生产系统上进行验证，更多地了解其行为，并为对更复杂的策略进行测试和研究奠定基础。

大型HDFS实例通常运行在跨多个机架上的服务器集群。不同机架上的两个节点是通过交换机实现通讯的，在大多数情况下，在同一机架上的服务器网络带宽要优于在不同机架上的服务器网络带宽。

NameNode通过Hadoop机架感知确定每个DataNode所属的机架ID。一个简单但不是最优的策略是将副本放在不同的机架上，这样可以防止当整个机架发生故障时丢失数据，并且在读取数据时可以使用不同机架的带宽。该策略把副本均匀地分布到集群中，使得在组件故障的时候能够容易地实现负载均衡。但是，这种策略会对写性能造成负影响，因为写的时候需要把数据块传输到多个不同的机架。

一般情况下，当副本数设置为3时，HDFS的副本放置策略是将一个副本放在本地节点，第二个副本放在同一机架的不同节点，第三个副本放在不同机架上。该策略减少机架间的写流量，提高了写性能。机架故障的几率远远低于节点的故障，该策略不会影响数据的可靠性和可用性保证。然而，当读数据的时候会减少网络带宽的使用，因为数据块仅存在两个不同的机架，而不是三个。文件的副本不是均匀地分布在不同的机架上，三分之一的副本在同一个节点，三分之二的副本在同一个机架，另外三分之一均匀地分布在其余的机架上。这种策略提高了写性能，而且不会影响数据的可靠性和读性能。
目前，这里描述的默认副本放置策略还处于开发之中。

副本选择

为了尽量减小全局带宽的消耗和读延时，HDFS会尝试从离请求最近的副本读取数据。如果在同一个机架有请求数据的副本，就直接读取，如果HDFS集群是部署在多个数据中心，那么会优先读取本地数据中心的副本，而不是远程的副本。
安全模式

在HDFS启动的时候，NameNode会进入一个叫安全模式Safemode的特别状态，在此模式下，数据块还不会被复制。NameNode接收来自DataNodes的心跳和Blockreport信息，一份Blockreport包含该DataNode所有块的列表。每一个块有一个特定的最小复制数，当数据块的最小复制数被NameNode检查后，就认为是复制成功。当达到配置的块复制安全比例时（加上额外的30秒），NameNode就退出安全模式状态。然后，它会检测数据块的列表，把少于指定数量副本的数据块复制到其它的DataNodes。

HDFS这一部分主要有以下几个部分组成：

Client：客户端，系统使用者。

       1、文件切分。文件上传 HDFS 的时候，Client 将文件切分成 一个一个的Block，然后进行存储。

       2、与 NameNode 交互，获取文件的位置信息。

       3、与 DataNode 交互，读取或者写入数据。

       4、Client 提供一些命令来管理 HDFS，比如启动或者关闭HDFS。

       5、Client 可以通过一些命令来访问 HDFS。

NameNode：就是 master，它是一个主管、管理者。

       1、管理 HDFS 的名称空间。

       2、管理数据块（Block）映射信息

       3、配置副本策略

       4、处理客户端读写请求。

DataNode：就是Slave。NameNode 下达命令，DataNode 执行实际的操作。

       1、存储实际的数据块。

       2、执行数据块的读/写操作。

       3、定期向NameNode发送心跳信息，汇报本身及其所有的block信息，健康情况。

Secondary NameNode：并非 NameNode 的热备。当NameNode 挂掉的时候，它并不能马上替换 NameNode 并提供服务。注意：在hadoop 2.x 版本，当启用 hdfs ha 时，将没有这一角色

       1、辅助 NameNode，分担其工作量。

       2、定期合并 fsimage和fsedits，并推送给NameNode。

       3、在紧急情况下，可辅助恢复 NameNode。

热备份：b是a的热备份，如果a坏掉。那么b立即运行代替a的工作

冷备份：b是a的冷备份，如果a坏掉。那么b不能立即代替a工作。但是b上存储a的一些信息，减少a坏掉之后的损失
HDFS存储原理

下面以图解的方式讲述HDFS文件读写流程（两种方式体现）：
数据写入过程

客户端：发送读写请求；

Namenode：把控所有请求

Datenede：数据存储

一个客户端创建一个文件的请求是不会立即发送到NameNode的，事实上在开始时，HDFS客户端会先把数据缓存到本地的一个缓冲区。应用程序会把数据库重定向写入到此缓冲区，当缓冲区累积的数据（当客户端向HDFS中写入数据的时候，首先会读取Hadoop的配置项，获取数据块的大小（大文件会被分割成多个Block进行存储，一般为64或128MB）以及备份数（每个Block会在多个Datanode上存储多份副本，一般为3份），之后将数据写到本地临时文件（缓存）中）超过配置的一个数据块大小时，客户端才会连接NameNode。

找空闲，均衡

NameNode将文件名插入到文件系统层次结构中并为其分配存储空间，接着向客户端返回DataNode的标识和数据块的路径。然后客户端将数据块从本地缓冲区传输到指定的DataNode。

以流水线的方式写完，

客户端会从Namenode获取一个Datanode列表用于存放数据块（Datanaode列表列出了存储数据块的地址，并根据距离对他们进行了排序）。然后客户端开始向第一个Datanode传输数据，第一个 Datanode 一小部分一小部分地接收数据，将每一部分写入本地仓库，并同时传输该部分到列表中 第二个 Datanode 节点。第二个 Datanode 也是这样，一小部分一小部分地接收数据，写入本地 仓库，并同时传给第三个 Datanode 。最后，第三个 Datanode 接收数据并存储在本地。当完成该数据块的存储后，Datanode会向Namenode报告数据传输完成，Namenode通知客户端该数据块已成功存储并复制在HDFS中，客户端继续重复发送下个数据块，直至所有数据块传送完成。

这时客户端向Namenode报告所有数据块都已写入，请求关闭文件。Namenode关闭文件，数据写入完毕。

当关闭一个文件时，本地缓冲区剩余的未传输的数据会传输到DataNode中，客户端然后会告诉NameNode文件已经关闭。此时，NameNode会将文件创建操作提交到持久化存储。如果NameNode在文件关闭之前宕机，文件将会丢失。
读取数据

客户端发送请求：元数据信息：几个block，存放位置，编号，几个副本

客户端从HDFS中读取文件，首先向Namenode请求要读取文件的信息，Namenode递给客户端数据块列表，客户端知道了有多少个数据块需要下载，也清楚了储存每个数据块的Datanode位置，就会逐个下载所有数据块（在写数据过程中，数据存储已经按照客户端与DataNode节点之间的距离进行了排序，距客户端越近的DataNode节点被放在最前面，客户端会优先从本地读取该数据块）

HDFS的文件写入原理，主要包括以下几个步骤：

1，客户端通过调用DistributedFileSystem的create()方法创建新文件；

2，DistributedFileSystem通过RPC调用NameNode去创建一个没有Blocks关联的新文件，创建前NameNode会做各种校验，比如文件是否存在、客户端有无权限去创建等。如果校验通过，NameNode会为创建新文件记录一条记录，否则就会抛出IO异常；

前两步结束后会返回FSDataOutputStream的对象，和读文件的时候相似，

3，FSDataOutputStream被封装成DFSOutputStream，DFSOutputStream可以协调NameNode和Datanode。客户端开始写数据到DFSOutputStream，DFSOutputStream会把数据切成一个个小的数据包，并写入内部队列称为“数据队列”(Data Queue)；

4，DataStreamer会去处理接受Data Queue，它先问询NameNode这个新的Block最适合存储的在哪几个DataNode里，比如重复数是3，那么就找到3个最适合的DataNode，把他们排成一个pipeline.DataStreamer把Packet按队列输出到管道的第一个Datanode中，第一个DataNode又把Packet输出到第二个DataNode中，以此类推；

5，DFSOutputStream还有一个对列叫Ack Quene，也是有Packet组成，等待DataNode的收到响应，当Pipeline中的所有DataNode都表示已经收到的时候，这时Akc Quene才会把对应的Packet包移除掉；

6，客户端完成写数据后调用close()方法关闭写入流；

7，DataStreamer把剩余的包都刷到Pipeline里然后等待Ack信息，收到最后一个Ack后，通知NameNode把文件标示为已完成。

HDFS的文件读取原理，主要包括以下几个步骤：

1，客户端通过调用FileSystem对象的open()方法来打开希望读取的文件，对于HDFS来说，这个对象是分布文件系统的一个实例；

2，DistributedFileSystem通过使用RPC（远程过程调用）来调用NameNode以确定文件起始块的位置，同一Block按照重复数会返回多个位置，这些位置按照Hadoop集群拓扑结构排序，距离客户端近的排在前面；

3，前两步会返回一个FSDataInputStream对象，该对象会被封装成DFSInputStream对象，DFSInputStream可以方便的管理datanode和namenode数据流，客户端对这个输入流调用read()方法；

4， 存储着文件起始块的DataNode地址的DFSInputStream随即连接距离最近的DataNode，通过对数据流反复调用read()方法，可以将数据从DataNode传输到客户端；

5，到达块的末端时，DFSInputStream会关闭与该DataNode的连接，然后寻找下一个块的最佳DataNode，这些操作对客户端来说是透明的，客户端的角度看来只是读一个持续不断的流；

6，一旦客户端完成读取，就对FSDataInputStream调用close()方法关闭文件读取。
文件系统元数据的持久化

HDFS的命名空间是由命名节点NameNode来存储的。NameNode使用了一个叫EditLog的事务日志来持续记录文件系统元数据的每一次更改，例如在HDFS创建一个新的文件，NameNode会在EditLog里面插入一条这样的记录。类似地，修改文件的复制因子也会在EditLog里面插入一条记录。NameNode使用本地服务器操作文件系统中的文件来存储EditLog。整个文件系统的命名空间，包括数据块的映射和文件系统的属性配置，都保存在一个叫FsImage的文件里，这个FsImage文件也是保存在NameNode的本地文件系统里。

NameNode将整个文件系统的命名空间和文件Blockmap的镜像保存在内存中。关键的元数据被设计成紧凑存储的，这使得一个有4GB内存的NameNode足以支持处理大量的文件和目录。当NameNode启动时，它从磁盘读取FsImage和EditLog文件，将EditLog的所有事务应用于FsImage的内存中，然后将这个新版本的FsImage刷新到磁盘中。因为事务已经被持久化到FsImage中，所以可以截去旧的EditLog，这个过程叫做检查点checkpoint。在当前实现中，checkpoint仅在NameNode启动时发生，而周期性的checkpoint功能正在实现中。

DataNode将HDFS的数据以文件形式存储到本地的文件系统中，而它在存储文件的时候又会将数据分成多个数据块分别存储在不同的文件中。DataNode不会将所有的数据块文件都存放到同一个目录中，而是它使用启发式方法来确定每个目录的最佳文件数，并适当地创建子目录。在本地同一个目录下创建所有的数据块文件不是最优的，因为本地文件系统可能无法在单个目录中有效地支持大量文件。当DataNode启动的时候，它将扫描它的本地文件系统，生成与这些本地文件相对应的所有数据块的列表，并将此列表发给NameNode，这个列表被称为Blockreport。
 通信协议

所有HDFS的通信协议都是构建在TCP/IP协议之上。一个客户端和NameNode的一个可配置的TCP端口建立连接使用的是客户端协议Client Protocol，DataNodes和NameNode通讯使用的是数据节点协议DataNode Protocol，远程程序调用(RPC)抽象封装了Client Protocol和DataNode Protocol。按照设计，NameNode从不发起任何RPC。相反，它只会响应DataNodes或客户端发出的RPC请求。
稳健性

HDFS设计的主要目标是即使出现故障的情况下也能够可靠地存储数据。三种常见的故障类型是NameNode故障，DataNode故障和网络分裂。

 数据磁盘故障，心跳和重新复制数据

每一个DataNode会周期性地向NameNode发送一个心跳包。网络分裂出现时可能会造成部分DataNodes和NameNode断开连接，NameNode会根据丢失心跳包检测到此类情况。NameNode会标记那些丢失心跳包的DataNodes为dead状态，并且不会再向其转发任何新的IO请求，而它们存储的数据将不再可用。DataNode被标记为dead可能会导致某些数据块的复制因子低于其指定值。NameNode会不断地跟踪哪些数据块需要复制，并在必要时启动数据复制。重新复制数据的必要性有很多，例如：DataNode故障，副本损坏，DataNode所在的磁盘损坏，或者文件的复制因子增加。
为了避免由于DataNodes状态摆动引起的数据频繁复制，标记DataNodes为dead状态的超时时间默认被设为保守的10分钟以上，计算公式为：
2 * dfs.namenode.heartbeat.recheck-interval + 10 * dfs.heartbeat.interval = 2 * 300000(ms) + 10 * 3(s) = 10.5 minutes

集群数据重新均衡

HDFS的架构是与数据重新均衡方案兼容的。例如，如果一个DataNode的可用空间低于某个阀值，数据会自动从一个DataNode移动到另外一个DataNode；如果对某些文件的访问量突然剧增，数据会被额外地复制和集群的其它数据会被重新均衡。这些类型的数据重新均衡方案尚未实现。

数据完整性

从DataNode读取的数据块有可能会损坏，这可以是由于存储设备故障、网络故障、或者有bugs的代码而导致的。HDFS的客户端软件实现了checksum方法对文件内容进行校验，当客户端创建一个HDFS文件时，它会计算文件的每个数据块的校验和，并将这些校验和存储在同一命名空间中的独立隐藏文件中。当客户端读取文件时，它会验证从DataNode读取的数据与对应的校验和文件是否匹配。如果不匹配，则客户端会选择从另外一个有该数据副本的DataNode读取数据。

元数据磁盘故障

FsImage和EditLog是HDFS的核心数据结构，这些文件的损坏可以导致HDFS不能正常工作。因此，NameNode可以配置支持维护多个FsImage和EditLog的副本。对FsImage或EditLog的任何更新都会触发FsImage和EditLog副本的同步更新。多个FsImage和EditLog的副本同步更新会降低NameNode可以支持处理的每秒事务效率。然而，这些影响是可以接受的，因为尽管HDFS应用是数据密集型的，但它们不是元数据密集型的。当NameNode重新启动时，它会选择最近一致的FsImage和Editlog文件来使用。

 快照

快照支持保存在特定时刻的数据，此功能的一个用法是将损坏的HDFS实例回滚到之前的一个好的时间点。
如何数据访问

应用程序可以有很多方法访问HDFS，HDFS自带提供了一个FileSystem的Java API供应用程序使用，用C语言封装此Java API的接口和REST API也可用。另外，HTTP浏览器也可以用来浏览HDFS实例的文件。通过使用NFS网关，HDFS可以被加载到客户端的本地文件系统。（NFS（Network File System）即网络文件系统，是FreeBSD支持的文件系统中的一种，它允许网络中的计算机之间通过TCP/IP网络共享资源。在NFS的应用中，本地NFS的客户端应用可以透明地读写位于远端NFS服务器上的文件，就像访问本地文件一样。）

FS Shell

HDFS允许以文件和目录的形式组织管理用户数据，它提供了一个称为FS shell的命令行接口让用户与HDFS中的数据进行交互。该命令集的语法类似于用户已经熟悉的其它shell(例如bach、csh)。下面是一些例子：

    在根目录/下面创建一个foodir的子目录：  bin/hadoop fs -mkdir /foodir
    删除根目录/下面的foodir子目录：            bin/hadoop fs -rm -R /foodir
    查看文件/foodir/myfile.txt的内容：             bin/hadoop fs -cat /foodir/myfile.txt

FS shell是针对需要使用脚本语言与存储数据进行交互的应用程序设计的。

DFSAdmin
DFSAdmin命令集是用于管理HDFS集群，这些命令是仅供HDFS管理员使用。下面是一些例子：

    使集群进入安全模式：  bin/hdfs dfsadmin -safemode enter
    生成DataNodes列表： bin/hdfs dfsadmin -report
    刷新DataNode(s)：   bin/hdfs dfsadmin -refreshNodes

浏览器接口
典型的HDFS安装配置了Web服务器用于通过可配置的TCP端口访问HDFS的namespace，这允许用户通过Web浏览器访问HDFS的namespace和查看其文件的内容。
空间回收

如果回收站被启用，通过FS Shell删除的文件不会被立即删除，而是先移到一个垃圾目录(每个用户在/user/<username>/.Trash下都有自己的垃圾文件目录)，只要文件还保留在回收站就可以被快速的恢复。
最近被删除的文件会被移到/user/<username>/.Trash/Current目录，在配置的间隔内，HDFS会在/user/<username>/.Trash/<date>创建文件的checkpoints目录，把Current目录下面的文件都移到该checkpoints目录里面，并删除过期的checkpoints目录。详细请参考FS shell的expunge命令。
在回收站的文件过期后，NameNode会从HDFS的namespace删除该文件，关联的数据块空间会被释放。注意，在用户执行删除文件后和HDFS真正释放空间会有一点延时。

下面是如何使用FS Shell删除文件的一个例子：

    在delete目录下面创建2个目录test1和test2

    $ hadoop fs -mkdir -p delete/test1
    $ hadoop fs -mkdir -p delete/test2
    $ hadoop fs -ls delete/
    Found 2 items
    drwxr-xr-x   - hadoop hadoop          0 2015-05-08 12:39 delete/test1
    drwxr-xr-x   - hadoop hadoop          0 2015-05-08 12:40 delete/test2

    删除test1，执行后会提示test1移动到用户的.Trash/Current目录

    $ hadoop fs -rm -r delete/test1
    Moved: hdfs://localhost:8020/user/hadoop/delete/test1 to trash at: hdfs://localhost:8020/user/hadoop/.Trash/Current

    使用skipTrash参数直接删除test2

    $ hadoop fs -rm -r -skipTrash delete/test2
    Deleted delete/test2

    查看回收站，只有test1，而没有test2

    $ hadoop fs -ls .Trash/Current/user/hadoop/delete/
    Found 1 items
    drwxr-xr-x   - hadoop hadoop          0 2015-05-08 12:39 .Trash/Current/user/hadoop/delete/test1

减小复制因子
当减小文件复制因子时，NameNode会选择可以删除的多余副本。在下一个Heartbeat把此信息传输给DataNode，此DataNode然后删除相应的数据块并释放空间。需要再次注意的是，调用setReplication API后和释放空间会有一点延时。
2，YARN资源调度框架：
YARN产生背景

MapReduce1.x时所存在的问题：单点故障&节点压力大&不易扩展

可以看到，1.x时也是Master/Slave这种主从结构，在集群上的表现就是一个JobTracker带多个TaskTracker。

JobTracker：

JobTracker是一个后台服务进程，启动之后，会一直监听并接收来自各个TaskTracker发送的心跳信息，包括资源使用情况和任务运行情况等信息。

负责资源管理和作业调度

    作业控制：在hadoop中每个应用程序被表示成一个作业，每个作业又被分成多个任务，JobTracker的作业控制模块则负责作业的分解和状态监控。
    状态监控：主要包括TaskTracker状态监控、作业状态监控和任务状态监控。主要作用：容错和为任务调度提供决策依据。
    JobTracker只有一个，他负责了任务的信息采集整理，你就把它当做包工头，把这个和采用Master/Slave结构中的Master保持一致
    JobTracker 对应于 NameNode
    一般情况应该把JobTracker部署在单独的机器上

TaskTracker：1，定期向JobTracker汇报本节点的健康状况、资源使用情况以及作业执行情况。

                         2， 接收来自JobTracker的命令，例如启动任务或结束任务等。

MapTask

    自己开发的map任务交由该Task处理
    解析每条记录的数据，交给自己的map方法处理；
    将map的输出结果写到本地磁盘（有些作业只有map没有reduce==》HDFS）

ReduceTask

    将MapTask输出数据进行读取；
    按照数据进行分组传给我们自己编写的reduce方法处理；
    输出结果写到HDFS。

那么这种架构存在哪些问题呢：

    整个集群中只有一个JobTracker，就代表着会存在单点故障的情况
    JobTracker节点的压力很大，不仅要接收来自客户端的请求，还要接收大量TaskTracker节点的请求
    由于JobTracker是单节点，所以容易成为集群中的瓶颈，而且也不易域扩展
    JobTracker承载的职责过多，基本整个集群中的事情都是JobTracker来管理
    1.x版本的整个集群只支持MapReduce作业，其他例如Spark的作业就不支持了

各个资源不能共享
YARN架构：

YARN：不同计算框架可以共享同一个HDFS集群上的数据，享受整体的资源调度

YARN的基本思想是将资源管理和作业调度/监控的功能分解为单独的守护进程（守护进程(daemon)是一类在后台运行的特殊进程，用于执行特定的系统任务。很多守护进程在系统引导的时候启动，并且一直运行直到系统关闭。另一些只在需要的时候才启动，完成任务后就自动结束。）。 这个想法是有一个全局的ResourceManager（RM）和每个应用程序的ApplicationMaster（AM）。 应用程序可以是单个作业，也可以是DAG作业。

1. ResourceManager， 简称RM，ResourceManager是仲裁系统中所有应用程序之间资源的最终权威机构。（大管理员）

    整个集群同一时间提供服务的RM只有一个，它负责集群资源的统一管理和调度。
    处理客户端的请求，例如：提交作业或结束作业等。
    监控集群中的NM，一旦某个NM挂了，那么就需要将该NM上运行的任务告诉AM来如何进行处理。

ResourceManager主要有两个组件：Scheduler和ApplicationManager。

Scheduler是一个资源调度器，它主要负责协调集群中各个应用的资源分配，保障整个集群的运行效率。Scheduler的角色是一个纯调度器，它只负责调度Containers，不会关心应用程序监控及其运行状态等信息。同样，它也不能重启因应用失败或者硬件错误而运行失败的任务。Scheduler是一个可插拔的插件，它可以调度集群中的各种队列、应用等。在Hadoop的MapReduce框架中主要有两种Scheduler：Capacity Scheduler和Fair Scheduler。

参考:https://blog.csdn.net/suifeng3051/article/details/49508261

另一个组件ApplicationManager主要负责接收job的提交请求，为应用分配第一个Container来运行ApplicationMaster，还有就是负责监控ApplicationMaster，在遇到失败时重启ApplicationMaster运行的Container。

2. NodeManager， 简称NM，（执行者，小员工）

    整个集群中会有多个NM，它主要负责自己本身节点的资源管理和使用，
    定时向RM汇报本节点的资源使用情况。
    接收并处理来自RM的各种命令，例如：启动Container。
    NM还需要处理来自AM的命令，例如：AM会告诉NM需要启动多少个Container来跑task。
    单个节点的资源管理

NodeManager进程运行在集群中的节点上，每个节点都会有自己的NodeManager。NodeManager是一个slave服务：它负责接收ResourceManager的资源分配请求，分配具体的Container给应用。同时，它还负责监控并报告Container使用信息给ResourceManager。通过和ResourceManager配合，NodeManager负责整个Hadoop集群中的资源分配工作。ResourceManager是一个全局的进程，而NodeManager只是每个节点上的进程，管理这个节点上的资源分配和监控运行节点的健康状态。下面是NodeManager的具体任务列表：

接收ResourceManager的请求，分配Container给应用的某个任务
和ResourceManager交换信息以确保整个集群平稳运行。ResourceManager就是通过收集每个NodeManager的报告信息来追踪整个集群健康状态的，而NodeManager负责监控自身的健康状态。

管理每个Container的生命周期
管理每个节点上的日志
执行Yarn上面应用的一些额外的服务，比如MapReduce的shuffle过程

当一个节点启动时，它会向ResourceManager进行注册并告知ResourceManager自己有多少资源可用。在运行期，通过NodeManager和ResourceManager协同工作，这些信息会不断被更新并保障整个集群发挥出最佳状态。

NodeManager只负责管理自身的Container，它并不知道运行在它上面应用的信息。负责管理应用信息的组件是ApplicationMaster。

3. ApplicationMaster， 简称AM，应用级别（小管理员）

    每个应用程序都对应着一个AM。例如：MapReduce会对应一个、Spark会对应一个。它主要负责应用程序的管理，
    为应用程序向RM申请资源（Core、Memory），将资源分配给内部的task。
    AM需要与NM通信，以此来启动或停止task。遇到失败的任务还负责重启它。

4. Container，

    封装了CPU、Memory等资源的一个容器
    是一个任务运行环境的抽象。

Container是Yarn框架的计算单元，是具体执行应用task（如map task、reduce task）的基本单位。Container和集群节点的关系是：一个节点会运行多个Container，但一个Container不会跨节点。

一个Container就是一组分配的系统资源，现阶段只包含两种系统资源（之后可能会增加磁盘、网络等资源）：

    1. CPU core
    2. Memory in MB

既然一个Container指的是具体节点上的计算资源，这就意味着Container中必定含有计算资源的位置信息：计算资源位于哪个机架的哪台机器上。所以我们在请求某个Container时，其实是向某台机器发起的请求，请求的是这台机器上的CPU和内存资源。

任何一个job或application必须运行在一个或多个Container中，在Yarn框架中，ResourceManager只负责告诉ApplicationMaster哪些Containers可以用，ApplicationMaster还需要去找NodeManager请求分配具体的Container。

5. Client， 客户端，

    提交作业
    查询作业的运行进度
    结束作业。

小结：

在MR1中，JobTracker即负责job的监控，又负责系统资源的分配。

    在MR2中，资源的调度分配由ResourceManager专门进行管理，
    每个job或应用的管理、监控交由相应的分布在集群中的ApplicationMaster，如果某个ApplicationMaster失败，ResourceManager还可以重启它，这大大提高了集群的拓展性。

在MR1中，Hadoop架构只支持MapReduce类型的job，所以它不是一个通用的框架，因为Hadoop的JobTracker和TaskTracker组件都是专门针对MapReduce开发的，它们之间是深度耦合的。

    Yarn的出现解决了这个问题，关于Job或应用的管理都是由ApplicationMaster进程负责的，Yarn允许我们自己开发ApplicationMaster，我们可以为自己的应用开发自己的ApplicationMaster。这样每一个类型的应用都会对应一个ApplicationMaster，一个ApplicationMaster其实就是一个类库。这里要区分ApplicationMaster*类库和ApplicationMaster实例*，一个ApplicationMaster类库可以对应多个实例，就行java语言中的类和类的实例关系一样。总结来说就是，每种类型的应用都会对应着一个ApplicationMaster，每个类型的应用都可以启动多个ApplicationMaster实例。所以，在yarn中，是每个job都会对应一个ApplicationMaster而不是每类。

Resource Request和Container

Yarn的设计目标就是允许我们的各种应用以共享、安全、多租户的形式使用整个集群。并且，为了保证集群资源调度和数据访问的高效性，Yarn还必须能够感知整个集群拓扑结构。为了实现这些目标，ResourceManager的调度器Scheduler为应用程序的资源请求定义了一些灵活的协议，通过它就可以对运行在集群中的各个应用做更好的调度，因此，这就诞生了Resource Request和Container。

具体来讲，一个应用先向ApplicationMaster发送一个满足自己需求的资源请求，然后ApplicationMaster把这个资源请求以resource-request的形式发送给ResourceManager的Scheduler，Scheduler再在这个原始的resource-request中返回分配到的资源描述Container。每个ResourceRequest可看做一个可序列化Java对象，包含的字段信息如下：

    <resource-name, priority, resource-requirement, number-of-containers>
    <!--
    ?resource-name：资源名称，现阶段指的是资源所在的host和rack，后期可能还会支持虚拟机或者更复杂的网络结构
    ?priority：资源的优先级
    ?resource-requirement：资源的具体需求，现阶段指内存和cpu需求的数量
    ?number-of-containers：满足需求的Container的集合
    -->

number-of-containers中的Containers就是ResourceManager给ApplicationMaster分配资源的结果。Container就是授权给应用程序可以使用某个节点机器上CPU和内存的数量。

ApplicationMaster在得到这些Containers后，还需要与分配Container所在机器上的NodeManager交互来启动Container并运行相关任务。当然Container的分配是需要认证的，以防止ApplicationMaster自己去请求集群资源。
YARN执行流程：

一个作业提交上来后，先到ResourceManager，然后ResourceManager到任意一个结点上去请求一个ApplicationMaster，然后ApplicationMaster去为作业向ResourceManager申请资源，申请资源后通知对应的NodeManager，然后NodeManager启动container运行。

详细流程：

    客户端程序向ResourceManager提交应用并请求一个ApplicationMaster实例

    ResourceManager找到可以运行一个Container的NodeManager，并在这个Container中启动ApplicationMaster实例

    ApplicationMaster向ResourceManager进行注册，注册之后客户端就可以查询ResourceManager获得自己ApplicationMaster的详细信息，以后就可以和自己的ApplicationMaster直接交互了

    在平常的操作过程中，ApplicationMaster根据resource-request协议向ResourceManager发送resource-request请求

    当Container被成功分配之后，ApplicationMaster通过向NodeManager发送container-launch-specification信息来启动Container， container-launch-specification信息包含了能够让Container和ApplicationMaster交流所需要的资料

    应用程序的代码在启动的Container中运行，并把运行的进度、状态等信息通过application-specific协议发送给ApplicationMaster

    在应用程序运行期间，提交应用的客户端主动和ApplicationMaster交流获得应用的运行状态、进度更新等信息，交流的协议也是application-specific协议

    一但应用程序执行完成并且所有相关工作也已经完成，ApplicationMaster向ResourceManager取消注册然后关闭，用到所有的Container也归还给系统。

Yarn 内存分配管理机制及相关参数配置

参考：https://blog.csdn.net/suifeng3051/article/details/48135521
3，MapReduce：分布式处理框架

优点：海量数据离线处理&易开发&易运行

缺点：不满足实时流式计算。它是离线处理，输入级都是固定的，是静态的，动态没办法识别：

多个应用程序存在依赖关系，dag不擅长处理。

横向扩展结点，spark：易开发，易运行。

Hadoop核心之MapReduce是一个软件框架，基于该框架能够容易地编写应用程序，这些应用程序能够运行在由上千个商用机器组成的大集群上，并以一种可靠的，具有容错能力的方式并行地处理上TB级别的海量数据集。这个定义里面有着这些关键词，一是软件框架，二是并行处理，三是可靠且容错，四是大规模集群，五是海量数据集。因此，对于MapReduce，可以简洁地认为，它是一个软件框架，海量数据是它的“菜”，它在大规模集群上以一种可靠且容错的方式并行地“烹饪这道菜”。

MapReduce擅长处理大数据，它为什么具有这种能力呢？这可由MapReduce的设计思想发觉。MapReduce的思想就是“分而治之”。Mapper负责“分”，即把复杂的任务分解为若干个“简单的任务”来处理。“简单的任务”包含三层含义：一是数据或计算的规模相对原任务要大大缩小；二是就近计算原则，即任务会分配到存放着所需数据的节点上进行计算；三是这些小任务可以并行计算，彼此间几乎没有依赖关系。Reducer负责对map阶段的结果进行汇总。至于需要多少个Reducer，用户可以根据具体问题，通过在mapred-site.xml配置文件里设置参数mapred.reduce.tasks的值，缺省值为1。
通过wordcount词频统计分析工作流程：

将作业拆分成Map阶段和Reduce阶段

Map阶段：Map Tasks

Reduce阶段：Reduce Tasks
MapReduce作业的输入和输出类型：

（输入）<k1，v1> - > map - > <k2，v2> - > combine - > <k2，v2> - > reduce - > <k3，v3>（输出）

接下来是总说，下面举例细说
map task

程序会根据InputFormat将输入文件分割成splits，每个split会作为一个map task的输入，每个map task会有一个内存缓冲区,输入数据经过map阶段处理后的中间结果会写入内存缓冲区，并且决定数据写入到哪个partitioner，当写入的数据到达内存缓冲区的的阀值(默认是0.8)，会启动一个线程将内存中的数据溢写入磁盘，同时不影响map中间结果继续写入缓冲区。在溢写过程中，MapReduce框架会对key进行排序，如果中间结果比较大，会形成多个溢写文件，最后的缓冲区数据也会全部溢写入磁盘形成一个溢写文件(最少有一个溢写文件)，如果是多个溢写文件，则最后合并所有的溢写文件为一个文件。

split被送入map task后，程序库决定数据结果数据属于哪个partitioner，写入到内存缓冲区，到达阀值，开启溢写过程，进行key排序，如果有combiner步骤，则会对相同的key做归并处理，最终多个溢写文件合并为一个文件。

MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。

多个map task形成的最终文件的对应partitioner会被对应的reduce task拉取至内存缓冲区，对可能形成多个溢写文件合并，最终作为resuce task的数据输入 。
reduce task

当所有的map task完成后，每个map task会形成一个最终文件，并且该文件按区划分。reduce任务启动之前，一个map task完成后，就会启动线程来拉取map结果数据到相应的reduce task，不断地合并数据，为reduce的数据输入做准备，当所有的map tesk完成后，数据也拉取合并完毕后，reduce task 启动，最终将输出输出结果存入HDFS上。
Map、Reduce任务中Shuffle和排序的过程

Shuffle的过程：描述数据从map task输出到reduce task输入的这段过程。

 Shuffle的正常意思是洗牌或弄乱，可能大家更熟悉的是Java API里的Collections.shuffle(List)方法，它会随机地打乱参数list里的元素顺序。

参考：http://langyu.iteye.com/blog/992916

           http://weixiaolu.iteye.com/blog/1474172
Map端：

1，在map task执行时，它的输入数据来源于HDFS的block，当然在MapReduce概念中，map task只读取split。Split与block的对应关系可能是多对一，默认是一对一。在WordCount例子里，假设map的输入数据都是像“aaa”这样的字符串。

2， 在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对： key是“aaa”， value是数值1。在我们的例子中，“aaa”经过Partitioner后返回0，也就是这对值应当交由第一个reducer来处理。接下来，需要将数据写入内存缓冲区中，缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。

3，当写入的数据到达内存缓冲区的的阀值(默认是0.8)，会启动一个线程将内存中的数据溢写入磁盘，同时不影响map中间结果继续写入缓冲区。在溢写过程中，MapReduce框架会对key进行排序，

    溢写过程一个很重要的细节在于，如果有很多个key/value对需要发送到某个reduce端去，那么需要将这些key/value值拼接到一块，减少与partition相关的索引记录。 在针对每个reduce端而合并数据时，有些数据可能像这样：“aaa”/1， “aaa”/1。对于WordCount例子，就是简单地统计单词出现的次数，如果在同一个map task的结果中有很多个像“aaa”一样出现多次的key，我们就应该把它们的值合并到一块，这个过程叫combine。

4，如果中间结果比较大，会形成多个溢写文件，最后的缓冲区数据也会全部溢写入磁盘形成一个溢写文件(最少有一个溢写文件)，如果是多个溢写文件，则最后合并所有的溢写文件为一个文件。这个过程就叫做Merge。

    Merge是怎样的？如前面的例子，“aaa”从某个map task读取过来时值是5，从另外一个map 读取时值是8，因为它们有相同的key，所以得merge成group。什么是group。对于“aaa”就是像这样的：{“aaa”, [5, 8, 2, …]}，数组中的值就是从不同溢写文件中读取出来的，然后再把这些值加起来。请注意，因为merge是将多个溢写文件合并到一个文件，所以可能也有相同的key存在，在这个过程中如果client设置过Combiner，也会使用Combiner来合并相同的key。

map端的所有工作都已结束，最终生成的这个文件也存放在TaskTracker够得着的某个本地目录内。每个reduce task不断地通过RPC从JobTracker那里获取map task是否完成的信息，如果reduce task得到通知，获知某台TaskTracker上的map task执行完成，Shuffle的后半段过程开始启动。

到这里，map端就分析完了。那到底什么是Shuffle呢？Shuffle的中文意思是“洗牌”，如果我们这样看：一个map产生的数据，结果通过hash过程分区却分配给了不同的reduce任务，是不是一个对数据洗牌的过程呢。
Reduce端：

1．Reduce会接收到不同map任务传来的数据，并且每个map传来的数据都是有序的。如果reduce端接受的数据量相当小，则直接存储在内存中（缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制，表示用作此用途的堆空间的百分比），如果数据量超过了该缓冲区大小的一定比例（由mapred.job.shuffle.merge.percent决定），则对数据合并后溢写到磁盘中。
2．随着溢写文件的增多，后台线程会将它们合并成一个更大的有序的文件，这样做是为了给后面的合并节省时间。其实不管在map端还是reduce端，MapReduce都是反复地执行排序，合并操作，现在终于明白了有些人为什么会说：排序是hadoop的灵魂。
3．合并的过程中会产生许多的中间文件（写入磁盘了），但MapReduce会让写入磁盘的数据尽可能地少，并且最后一次合并的结果并没有写入磁盘，而是直接输入到reduce函数。
Reduce 端的Shuffle细节

1.Copy过程，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求map task所在的TaskTracker获取map task的输出文件。因为map task早已结束，这些文件就归TaskTracker管理在本地磁盘中。

2.Merge阶段。这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置，因为Shuffle阶段Reducer不运行，所以应该把绝大部分的内存都给Shuffle用。这里需要强调的是，merge有三种形式：1)内存到内存  2)内存到磁盘  3)磁盘到磁盘。默认情况下第一种形式不启用。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。

3.Reducer的输入文件。不断地merge后，最后会生成一个“最终文件”。为什么加引号？因为这个文件可能存在于磁盘上，也可能存在于内存中。对我们来说，当然希望它存放于内存中，直接作为Reducer的输入，但默认情况下，这个文件是存放于磁盘中的。当Reducer的输入文件已定，整个Shuffle才最终结束。然后就是Reducer执行，把结果放到HDFS上。
MapReduce编程主要组件

InputFormat类：分割成多个splits和每行怎么解析。   

Mapper类：对输入的每对<key,value>生成中间结果。

Combiner类：在map端，对相同的key进行合并。

Partitioner类：在shuffle过程中，将按照key值将中间结果分为R份，每一份都由一个reduce去完成。

Reducer类：对所有的map中间结果，进行合并。

OutputFormat类：负责输出结果格式。

文件输入格式InputFormat

1）定义了数据文件如何分割和读取

2）InputFile提供了以下一些功能

选择文件或者其它对象，用来作为输入

定义InputSplits，将一个文件分开成为任务

为RecordReader提供一个工厂，用来读取这个文件

3）有一个抽象的类FileInputFormat，所有的输入格式类都从这个类继承这个类的功能以及特性。当启动一个Hadoop任务的时候，一个输入文件所在的目录被输入到FileInputFormat对象中。FileInputFormat从这个目录中读取所有文件。然后FileInputFormat将这些文件分割为一个或者多个InputSplits。

4）通过在JobConf对象上设置JobConf.setInputFormat设置文件输入的格式

输入数据分块InputSplits

1）InputSplit定义了输入到单个Map任务的输入数据

2）一个MapReduce程序被统称为一个Job，可能有上百个任务构成

3）InputSplit将文件分为64MB的大小

配置文件hadoop-site.xml中的mapred.min.split.size参数控制这个大小

4）mapred.tasktracker.map.taks.maximum用来控制某一个节点上所有map任务的最大数目

数据记录读入RecordReader（RR）

1）InputSplit定义了一项工作的大小，但是没有定义如何读取数据

2）RecordReader实际上定义了如何从数据上转化为一个(key,value)对的详细方法，并将数据输出到Mapper类中

3）TextInputFormat提供了LineRecordReader

Mapper

1）每一个Mapper类的实例生成了一个Java进程（在某一个InputSplit上执行）

2）有两个额外的参数OutputCollector以及Reporter，前者用来收集中间结果，后者用来获得环境参数以及设置当前执行的状态。

3）现在用Mapper.Context提供给每一个Mapper函数，用来提供上面两个对象的功能

Combiner

1）合并相同key的键值对，减少partitioner时候的数据通信开销

2）conf.setCombinerClass(Reduce.class)

Partitioner & Shuffle

在Map工作完成之后，每一个 Map函数会将结果传到对应的Reducer所在的节点，此时，用户可以提供一个Partitioner类，用来决定一个给定的(key,value)对传输的具体位置

Sort

传输到每一个节点上的所有的Reduce函数接收到得Key,value对会被Hadoop自动排序（即Map生成的结果传送到某一个节点的时候，会被自动排序）

Reducer

1）做用户定义的Reduce操作

2）最新的编程接口是Reducer.Context

文件输出格式OutputFormat

1）说明

写入到HDFS的所有OutputFormat都继承自FileOutputFormat

每一个Reducer都写一个文件到一个共同的输出目录，文件名是part-nnnnn，其中nnnnn是与每一个reducer相关的一个号（partition id）

FileOutputFormat.setOutputPath()

JobConf.setOutputFormat()

2）接口定义

RecordWriter

TextOutputFormat实现了缺省的LineRecordWriter，以”key／value”形式输出一行结果。
4，HIVE（数据仓库）

解决含量结构化日志，数据统计问题的，类似SQL语言。离线分析，SQL语句利用HIVE引擎转化。

Hive--Hive类似于SQL高级语言，用于运行存储在Hadoop上的查询语句，Hive让不熟悉MapReduce开发人员也能编写数据查询语句，然后这些语句被翻译为Hadoop上面的MapReduce任务。像Pig一样，Hive作为一个抽象层工具，吸引了很多熟悉SQL而不是Java编程的数据分析师。
5，RConnectors

解决统计分析的
6，Mahout：（数据挖掘算法库）；

Mahout是一个机器学习和数据挖掘库，它提供的MapReduce包含很多实现，包括聚类算法、回归测试、统计建模。通过使用 Apache Hadoop 库，可以将Mahout有效地扩展到云中。
7，Pig（数据流处理）

通过脚本的方式转成Map Reduce提交到集群运算，离线分析；它是MapReduce编程的复杂性的抽象。Pig平台包括运行环境和用于分析Hadoop数据集的脚本语言(Pig Latin)，其编译器将Pig Latin翻译成MapReduce程序序列。
8，Oozie 工作流调度引擎

Oozie是一个可扩展的工作体系，集成于Hadoop的堆栈，用于协调多个MapReduce作业的执行。它能够管理一个复杂的系统，基于外部事件来执行，外部事件包括数据的定时和数据的出现。

一层一层去审批，有一定的依赖关系，配置好，运行。
9，Zookeeper（分布式协作服务）

分布式的协调服务，管理框架，如解决单点故障切换问题；用于 Hadoop 的分布式协调服务。Hadoop 的许多组件依赖于 Zookeeper，它运行在计算机集群中，用于管理Hadoop集群
10，Flume（日志收集工具）

日志收集框架，应用场景：日志的统计分析操作。Flume提供了分布式、可靠、高效的服务，用于收集、汇总大数据，并将单台计算机的大量数据转移到HDFS。它基于一个简单而灵活的架构，并提供了数据流的流。它利用简单的可扩展的数据模型，将企业中多台计算机上的数据转移到Hadoop中。
11，Sqoop（数据库ETL工具）

与传统数据库（关系型数据库）的数据传输与交换。Sqoop是一个连接工具，用于在关系数据库、数据仓库和Hadoop之间转移数据。Sqoop利用数据库技术描述架构，进行数据的导入/导出；利用MapReduce实现并行化运行和容错技术。
12，Hbase（实时分布式数据库）：

链式的存储，结构化数据的可伸缩可扩展高性能面向链的数据库。非常大，快速查询，秒级别查询上亿，可进行实时查询HBase 是一个建立在 HDFS 之上，面向列的 NoSQL 数据库，用于快速读/写大量数据，HBase 使用 Zookeeper 进行管理。