Hadoop 底层原理介绍

1 概述

Apache Hadoop是一个软件框架，可在具有数千个节点和PB级数据的大型集群上进行分布式处理。

Hadoop主要包含四个项目：Hadoop Common、Hadoop分布式文件系统(HDFS)、YARN(Yet Another Resource Negoiator)、MapReduce。

HDFS用于存储数据
MapReduce用于处理数据
YARN用来管理集群的资源(CPU和内存)及支持Hadoop的公共实用程序

Hadoop带来三个组件：

可靠的分布式数据存储框架、
用于并行处理数据的框架
用于集群资源管理的框架

Hadoop1.0 和 Hadoop2.0 结构对比：

Hadoop2.0 主要改进：

1、YARN实现资源调度和管理，使Hadoop2.0可以运行更多种类的计算框架，如Spark、Giraph等

2、实现了NameNode的HA方案，即同时有2个NameNode（一个Active另一个Standby），如果ActiveNameNode挂掉的话，另一个NameNode会转入Active状态提供服务，保证了整个集群的高可用。

3、实现了HDFS federation，由于元数据放在NameNode的内存当中，内存限制了整个集群的规模，通过HDFS federation使多个NameNode组成一个联邦共同管理DataNode，这样就可以扩大集群规模。

4、Hadoop RPC序列化扩展性好，通过将数据类型模块从RPC中独立出来，成为一个独立的可插拔模块。

RPC：RPC 的主要功能目标是让构建分布式计算（应用）更容易，在提供强大的远程调用能力时不损失本地调用的语义简洁性。为实现该目标，RPC 框架需提供一种透明调用机制让使用者不必显式的区分本地调用和远程调用。

HA(High Available)：双机集群系统的简称，指高可用性集群，是保证业务连续性的有效解决方案，一般有两个或两个以上的节点，且分为活动节点及备用节点。通常把正在执行业务的称为活动节点，而作为活动节点的一个备份的则称为备用节点。当活动节点出现问题，导致正在运行的业务（任务）不能正常运行时，备用节点此时就会侦测到，并立即接续活动节点来执行业务。从而实现业务的不中断或短暂中断。

2 HDFS概述

HDFS是一种分布式文件系统，在低价通用硬件的大型集群上提供高扩展性和可靠性。具有高容错的特点。

HDFS文件会划分成数据块，分布在集群中，每个区块大小通常128MB。每个区块会被复制3份，以便故障处理。

区块的元数据(文件名、区块、位置、创建的日期和大小)会存储在管理结点中。HDFS具有较大的区块大小，从而减少了读取完整文件所需的磁盘寻道次数。

主要特点：

1、支持超大文件：支持TB级的数据文件。

2、检测和快速应对硬件故障：HDFS的检测和冗余机制很好克服了大量通用硬件平台上的硬件故障问题。

3、高吞吐量：批量处理数据。

4、简化一致性模型：一次写入多次读取的文件处理模型有利于提高吞吐量。

构成：NameNode保存HDFS的元数据；DataNode将HDFS数据以文件形式存储在本地文件系统上，并不知道有关HDFS文件的信息。

数据块：是HDFS的文件存储处理单元，默认128MB，可根据业务配置。数据块的存在，使HDFS可保存比存储节点单一磁盘大的文件，简化了存储管理，方便容错，有利于数据复制。

3 HDFS读写流程

读文件流程：

1. client使用open()打开文件

2. 分布式文件系统用RPC调用数据节点，得到文件数据块信息

3. 对每个数据块，NameNode返回保存数据块的数据节点的地址

4. 分布式文件系统返回FSDataInputStream给client，用来读取数据

5.client调用FSDataInputStream的read()开始读取数据

6. FSDataInputStream连接保存此文件第一个数据块的最近数据节点

7. data从数据节点读到 client

8. 当此数据块读取完毕，FSDataInputStream关闭和此数据节点的连接，然后连接此文件下一数据块的最近数据节点

9. 当client读取数据完毕，调用FSDataInputStream的close()函数

10. 读取数据过程中。若client与数据节点通信出现错误，则尝试连接包含此数据块的下一个数据节点。失败的数据节点将被记录，以后不再连接。

流程图如下：

写文件流程：

1.client调用create()创建文件

2. 分布式文件系统用RPC调用NameNode节点，在文件系统的命名空间中创建一个新的文件

3. NameNode节点首先确定文件是否存在，并且client是否有创建文件的权限，然后创建新文件

4. 分布式文件系统返回FSDataOutputStream给client用于写数据

5. client开始写入数据，FSDataOutputStream将数据分块，写入data queue

6. data queue 由 DataStreamer 读取，并通知NameNode节点分配数据节点用来存储数据块（每块默认复制3块），分配的数据节点放在一个pipeline里

7. DataStreamer 将数据块写入pipeline中第一个数据节点，第一个数据节点把数据块发给第二个，第二个发给第三个

8. FSDataOutputStream 为发出去的数据块保存ack queue，等待pipeline中数据节点告知数据已写入成功

9. 若数据节点写入过程中失败，则进行以下操作：

（1）关闭pipeline并把ack queue中数据块放入data queue的开始

（2）当前数据块在已写入的数据节点中被NameNode结点赋予新的标识，错误节点重启后察觉其数据块过时而被删除

（3）失败数据节点从pipeline移除，另外的数据块则写入pipeline中另外的数据节点

（4）NameNode结点被通知此数据块复制块数不足，从而创建第三份备份

10. 客户端结束写入数据，调用close（）将所有数据块写入pipeline中数据节点，等待ack queue返回成功，最后通知NameNode结点写入完毕。

流程图如下：

4 YARN原理介绍

4.1 YARN产生背景

1. 扩展性差：JobTracker同时兼备了资源管理和作业控制两个功能，这是整个系统的最大瓶颈，它严重制约了整个集群的扩展性

2. JobTracker存在单点故障，JobTracker出现问题将导致整个集群不可用

3. 资源利用率低：资源无法在多个任务间共享或合理分配，导致无法有效利用各种资源。

4. 无法支持多种计算框架：Hadoop 1.0只支持MapReduce这种离线批处理计算模式，而无法支持内存计算、流式计算、迭代式计算等。

注：内存计算：随着大容量内存技术的兴起，专家开始提出在初始阶段就把数据全部加载到内存中，而后可直接把数据从内存中调取出来，再由处理器进行计算。这样可以省去外存与内存之间的数据调入／调出过程，从而大大提升计算速度。

流式计算：利用分布式的思想和方法，对海量“流”式数据进行实时处理，源自业务对海量数据，在“时效”的价值上的挖掘诉求。

4.2 YARN基本架构

YARN是Hadoop2.0的资源管理器。它是一个通用的资源管理系统，可为上层应用提供统一的资源管理和调度。

YARN的基本设计思想是将Hadoop 1.0中的JobTracker拆分成了两个独立的服务：

一个全局的资源管理器ResourceManager：负责整个系统的资源管理和分配

每个应用程序特有的ApplicationMaster：负责单个应用程序的管理

基本架构如下图：

YARN总体上是Master/Slave结构，ResourceManager为Master，NodeManager为Slave，通过HA方案实现ResourceManager高可用。ResourceManager负责对各个NodeManager上的资源进行统一管理和调度。当用户提交一个应用程序时，需要提供一个用以跟踪和管理这个程序的ApplicationMaster，它负责向ResourceManager申请资源，并要求NodeManger启动可以占用一定资源的任务。由于不同的ApplicationMaster被分布到不同的节点上，因此它们之间不会相互影响。

Master/Slave结构：一个HDFS集群包含一个单独的Master节点和多个Slave节点服务器，这里的一个单独的Master节点的含义是HDFS系统中只存在一个逻辑上的Master组件。一个逻辑的Master节点可以包括两台物理主机，即两台Master服务器、多台Slave服务器。一台Master服务器组成单NameNode集群，两台Master服务器组成双NameNode集群，并且同时被多个客户端访问。

ResourceManager：全局资源管理器，负责真个系统的资源管理和分配，主要由调度器和应用程序管理器两个组件构成

调度器：根据容量、队列等限制条件，将系统中的资源分配给各个正在运行的应用程序。调度器仅根据应用程序的资源需求进行资源分配，而资源分配单位用一个抽象概念“资源容器”（简称Container）表示，Container是一个动态资源分配单位，它将内存、CPU、磁盘、网络等资源封装在一起，从而限定每个任务使用的资源量。

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

ApplicationMaster：用户提交的每个应用程序均包含1个ApplicationMaster，主要功能包括与ResourceManager调度器协商以获取资源、将得到的任务进一步分配给内部的任务、与NodeManager通信以启动/停止任务、监控所有任务运行状态并在任务运行失败时重新为任务申请资源以重启任务等。

NodeManager：它是每个节点上的资源和任务管理器，它不仅定时向ResourceManager汇报本节点上的资源使用情况和各个Container的运行状态，还接收并处理来自ApplicationMaster的Container启动/停止等各种请求。

Container：它是YARN中的资源抽象，它封装了某个节点上的多维度资源，如内存、CPU、磁盘、网络等，当ApplicationMaster向ResourceManager申请资源时，返回的资源便是用Container表示的。YARN会为每个任务分配一个Container，且该任务只能使用该Container中描述的资源。

4.3 YARN工作流程

1. 用户向YARN提交应用程序，包括用户程序、ApplicationMaster程序、ApplicationMaster启动命令等

2. ResourceManager为应用程序分配第一个Container，并与对应的NodeManager通信，要求它在这个Container中启动应用程序的ApplicationMaster。

3. ApplicationMaster首先向ResourceManager注册，这样用户可以通过ResourceManager查看程序运行状态，然后ApplicationMaster为各个任务申请资源，并监控它们的运行状态，直到运行结束，即重复步骤4-7

4.ApplicationMaster采用轮询的方式通过RPC协议向ResourceManager申请和领取资源

5.一旦ApplicationMaster成功申请到资源，便开始与对应的NodeManager通信，要求启动任务

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

7. 各个任务通过某个RPC协议向ApplicationMaster汇报自己的状态和进度，使ApplicationMaster能够随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。在应用程序运行过程中，用户可随时通过RPC向ApplicationMaster查询应用程序的当前运行状态

8.应用程序运行完成后，ApplicationMaster通过RPC协议向ResourceManager注销并关闭自己

5 MapReduce原理介绍

5.1 MapReduce介绍

MapReduce是由Google公司研究提出的一种面向大规模数据处理的并行计算模型和方法，是Hadoop面向大数据并行处理的计算模型、框架和平台。

MapReduce执行流包括input、map、shuffle、reduce和output共5个过程，如下图所示：

5.2 MapReduce原型原理：

YARN框架下MapReduce工作流程图：

1. 客户端向集群提交作业

2. Job从ResouceManager获取新的作业应用程序ID

3. 客户端检查作业的输出情况，计算分片，将作业jar包、配置、分片信息等作业资源复制到HDFS

4.Job向ResourceManager提交作业

5. ResourceManager接收到作业后，将作业请求传递给调度器，调度器根据作业信息为ResourceManager分配一个container，然后ResourceManager在NodeManager的管理下，在container中启动一个ApplicationMaster进程

6. ApplicationMaster对作业进行初始化，并保持对作业的跟踪，判断作业是否完成

7. ApplicationMaster根据存储在HDFS中的分片信息确定Map和Reduce的数量

8. ApplicationMaster为本次作业的Map和Reduce以轮询的方式向ResourceManager申请container

9. ApplicationMaster获取到container后，与NodeManager进行通信启动container

10. container从HDFS中获取作业的jar包、配置和分布式缓存文件等，将任务需要的资源本地化

11. container启动Map或Reduce任务

5.3 shuffle及排序

MapReduce的map端输出作为输入传给reduce端，并按键排序的过程称为shuffle。shuffle字面含义是洗牌，即将map产生的数据通过分区、排序等过程分配给不同reduce任务。MapReduce数据处理流程：

Map阶段：

1. 每个输入分片让一个map任务处理，默认情况下，以HDFS一个块的大小为一个分片。map输出结果暂存在一个环形内存缓冲区中。当该缓冲区快要溢出时，会在本地文件系统创建溢出文件，将该缓冲区中的数据写入这个文件。

2. 在写入磁盘之前，线程首先根据reduce任务的数目将数据划分为相同数目的分区，也就是一个reduce任务对应一个分区的数据。然后对每个分区中的数据进行排序，如果此时设置了Combiner，将排序后的结果进行combine操作，这样做可以有效减少磁盘IO和网络IO。

3. 当map任务输出最后一个记录时，可能会有很多的溢出文件，这时需要将这些文件合并。合并的过程中会不断地进行排序和combine操作，这样做是为了尽量减少每次写入磁盘的数据量和尽量减少下一复制阶段网络传输的数据量。最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量，这里可以将数据压缩。

4. 将分区中的数据拷贝给相对应的reduce任务。ApplicationMaster保存了整个作业的宏观信息，只要reduce任务向ApplicationMaster获取对应的map输出位置就可以了。

Reduce阶段：

1. Reduce会接收到不同map任务传来的数据，并且每个map传来的数据都是有序的。如果reduce接受的数据量相当小，则直接存储在内存中，如果数据量超过了该缓冲区大小的一定比例，则对数据合并后溢写到磁盘中

2. 随着溢写文件的增多，后台线程会将它们合并成一个更大的有序文件，这样做是为了给后面的合并节省时间。其实不管在map端还是reduce端，MapReduce都是反复地执行排序、合并操作，所以说排序是hadoop的灵魂

3.在合并的过程中会产生许多的中间文件（写入磁盘了），但MapReduce会让写入磁盘的数据尽可能地少，并且最后一次合并的结果并没有写入磁盘，而是直接输入到reduce函数