分布式计算框架Hadoop核心组件

Hadoop作为成熟的分布式计算框架在大数据生态领域已经使用多年，本文简要介绍Hadoop的核心组件MapReduce、YARN和HDFS，以加深了解。

---

1、Hadoop基本介绍

Hadoop是分布式计算框架，主要解决海量数据的存储和计算问题。Hadoop主要组件包括分布式文件系统HDFS、分布式离线并行计算框架MapReduce、作业调度与集群资源管理框架YARN。Hadoop生态系统一系列框架和组件如下：

2、MapReduce计算框架

Hadoop 1.0主要由两部分组成，分别是分布式文件系统HDFS和分布式计算框架Map/Reduce。其中分布式文件系统主要用于大规模数据的分布式存储，而MapReduce 则构建在分布式文件系统之上，对存储在分布式文件系统中的数据进行分布式计算。MapReduce作为一个分布式运算程序的编程框架，其核心功能是将用户编写的业务逻辑代码和自带默认组件整合一个完整的分布式运算程序，并发布到Hadoop集群上运行。

2.1 MapReduce架构

Hadoop MapReduce采用了Master/Slave（M/S）架构，它主要由以下几个组件组成：Client、JobTracker、 TaskTracker 和Task。

1）client

用户自定义的MapReduce程序通过Client提交到JobTracker端；同时，用户可通过Client提供的一些接口查看作业运行状态。在Hadoop内部用“作业”（Job）表示MapReduce程序，一个MapReduce程序可对应若干个作业，而每个作业会被分解成若干个Map/Reduce任务（Task）。

2）JobTracker

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

3）TaskTracker

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

4）Task

Task分为Map Task和Reduce Task两种，均由TaskTracker启动。从上一小节中我们知道，HDFS以固定大小的block为基本单位存储数据，而对于MapReduce而言，其处理单位是split。split是一个逻辑概念，它只包含一些元数据信息，比如数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。但需要注意的是，split的多少决定了Map Task的数目，因为每个split会交由一个Map Task处理。

2.2 MapReduce编程模型

MapReduce适用的应用场景具有一个共性：任务可以被分解为相互独立的子任务。基于该特点，MapReduce提供了其分布式编程的方法：

1. 迭代（iteration）。遍历输入数据，并将之解析成key/value对
2. 将输入key/value对映射（map）成另外一些key/value对
3. 依据key对中间数据进行分组（grouping）
4. 以组为单位对数据进行归约（reduce）
5. 迭代。将最终产生的key/value对保存到输出文件中

MapReduce 编程模型主要思想分为InputFormat、Split、Mapper、Shuffle、Reducer 和OutputFormat。

• InputFormat：主要用于描述输入数据的格式，将文件拆分为多个InputSplit，并由RecordReaders将InputSplit转换为标准的<key，value>键值对，作为map的输出；
• Split：对数据按行进行粗粒度切分，得到<Key,Value>型数据
• Map：细粒度切分，得到<Key,List>型数据；在环形缓冲区内对文件进行排序、分区操作，数据量大时会溢写到磁盘，缓冲区大小将决定着MR任务的性能，默认size为100M。此过程可以设置Combine任务，即将按照相同Key进行初步聚合
• Shuffle：将各个MapTask结果合并输出到Reduce，此过程数据的输出是一个Copy过程，此过程涉及到网络IO ，是一个耗时的过程，也是一个核心的过程。
• Reduce：对拆开的数据碎片进行合并，会涉及到Merge排序。
• OutputFormat：主要用于描述输出数据的格式，它能够将用户提供的key/value对写入特定格式的文件中

2.3 JobTracker内部实现

JobTracker是整个MapReduce计算框架中的主服务，负责整个集群的作业控制和资源管理。在Hadoop内部，每个应用程序表示为一个作业，每个作业又进一步分成多个任务。JobTracker的作用就是就是负责作业的分解以及状态监控，其中状态监控又包括TaskTracker状态监控、作业状态监控和任务状态监控。

2.3.1 状态监控

状态监控的一个重要目的是实现容错功能，借助监控信息，JobTracker实现全方位的容错机制，同时推测出运行缓慢的任务，并启动任务加快数据处理速度。

1. Job和Task运行状态监控：JobTracker为每个作业创建一个JobInProgress对象以跟踪其运行状态，同时也会将每个作业拆分成若干个任务，并为每个任务创建一个TaskInProgress对象以跟踪和监控其运行状态。
2. JobInProgress：用于监控和跟踪作业运行状态，并为调度器提供最底层的调度接口。其中维护了两种作业信息：静态信息和动态信息，静态信息在作业提交时已经确定好、动态信息随着作业的运行而动态变化。
3. TaskInProgress：维护Task运行过程中的全部信息

2.3.2 资源管理

Hadoop资源管理器由两部分组成：资源表示模型和资源分配模型，其中资源表示模型描述资源的组织方式，在Hadoop上使用slot组织各节点的资源；资源分配模型则决定如何将资源分配给各个作业，在Hadoop上通过调度器完成。Hadoop中引入slot概念，将各个节点上的资源等量的切分成若干份，每一份用一个slot表示。

在MapReduce框架中，由JobTracker实现资源调度。JobTracker不断接收各个TaskTracker周期性的发送过来的资源量和任务状态等信息，并综合考虑TaskTracker的数据分布、资源剩余量、作业优先级和作业提交时间等因素，为TaskTracker分配最合适的任务。

1. 客户端提交作业提交函数将程序提交到JobTracker端
2. JobTracker收到新作业后，通知任务调度器（TaskScheduler）对作业进行初始化
3. 某个TaskScheduler向JobTracker汇报心跳，其中包含剩余的slot数目和能否接收新任务等信息
4. 如果该TaskScheduler能够接收新任务，则JobTracker调用TaskScheduler对外函数asignTasks为该TaskScheduler分配新任务
5. TaskScheduler按照一定的调度策略为该TaskScheduler选择最合适的任务列表，并将该列表返回给JobTracker
6. JobTracker将任务列表以心跳应答的形式返回给对应的TaskTracker
7. TaskTracker收到心跳应答后，发现有需要启动的新任务，则直接启动该任务

2.4 TaskTrack内部实现

TaskTracker是Hadoop集群中运行于各个节点上的服务，是JobTracker和Tasks之间沟通的桥梁。TaskTracker主要实现两个功能：

1. 汇报心跳：周期性地将所在节点上的各种信息通过心跳机制汇报给JobTracker，包括节点健康状况、资源使用情况、任务执行进度、任务运行状态等
2. 执行命令：根据心跳信息和当前作业的运行情况为该TaskTracker下达命令，包括启动任务、提交任务、kill任务、kill作业和重新初始化

2.5 MapReduce运行过程

1. 客户端提交任务之前，（InputFormat）会根据配置策略将数据划分为切片(SpiltSize默认为blockSize 128M) ，每个切片都对应的提交给一个 MapTask (YARN 负责提交);
2. MapTask 执行任务，根据map函数，生成<K,V>对，将结果输出到环形缓存区，然后分区、排序、溢出；
3. Shuffle即将map结果划分到多个分区并分配给了多个reduce任务，此过程即为Shuffle。
4. Reduce拷贝map后分区的数据（fetch过程，默认5个线程执行拷贝），全部完成后执行合并操作。

2.5.1 Map Task

Map Task整体计算流程如图所示，分为5个阶段：

1. Read阶段：Map Task通过自定义的RecordReader，从输入InputSplit中解析出一个个key/value
2. Map阶段：将解析出的key-value交给自定义的map()函数，并产生一系列新的key-value
3. Collect阶段：在自定义的map()函数中，当数据处理完成后会调用OutPutCollectior.collect()输出结果。在该函数内部，会将生成的key/value分片，并写入环形的内存缓存区中
4. Spill阶段：当写入环形缓冲区的数据达到最大值的80%（默认环形缓冲区大小100M）会触发溢写操作spill，会将缓冲区的数据先按照partition进行排序再按照key进行排序，并将数据写入到磁盘中。
5. Combine阶段：当所有数据处理完毕后，Map Task将所有临时文件进行一次合并，最终只会生成一个文件，并为这个文件提供一个索引文件记录每个key对应数据的偏移量

2.5.2 Reduce Task

Reduce Task整体计算流程如图所示，共分为5个阶段：

1. Shuffle阶段：也称为copy阶段，Reduce Task从各个map task上远程拷贝一份数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中
2. Merge阶段：在远程拷贝数据的同时，Reduce Task启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多
3. Sort阶段：把分散的数据文件再次合并成一个大文件，再进行一次归并排序
4. Reduce阶段：将每组数据依次提交给自定义的reduce()函数处理
5. Write阶段：将计算结果写到HDFS中

2.5.3 Shuffle和Merge阶段分析

在Reduce Task中，Shuffle阶段和Merge阶段是并行进行的，可划分为三个子阶段。

1） 准备运行完成的Map Task列表

GetMapEventThread线程周期性的通过RPC从TaskTracker获取已完成的Map Task列表，并保存到映射表mapLocations中，其中保存了TaskTracker Host与已完成任务列表的映射关系。为防止出现网络热点，Reduce Task通过对所有TaskTracker Host进行混洗操作以打乱数据拷贝顺序，并将调整后 的Map Task输出数据位置保存到scheduledCopies中。

2） 远程拷贝数据

Reduce Task同时启动多个MapOutoutCopier线程，这些线程从scheduledCopies列表中获取Map Task输出位置，并通过HTTP get远程拷贝数据。对于获取的数据分片，如果大小超过一定的阈值，则存放在磁盘上，否则直接放到内存中。

3） 合并内存文件和磁盘文件

为了防止内存或者磁盘上的文件数据过多，Reduce Task启动了LocalFSMerger和InMemFSMergeThread两个线程分别对内存和磁盘上的文件进行合并。

3、YARN资源调度

YARN（Yet Another Resource Negotiator）是Hadoop 2.0中的资源管理系统，基本设计思想是将Hadoop 1.0中的JobTracker拆分成了两个独立的服务：一个全局的资源管理器ResourceManager和每个应用程序特有的ApplicationMaster。其中ResourceManager负责整个系统的资源管理和分配，而ApplicationMaster负责单个应用程序的管理。

3.1 YARN基本组成结构

YARN总体上仍然是Master/Slave架构，其中Resource Manager为master负责对各个NodeManager上的资源进行统一管理和调度、Node Manager为slave是每个节点上的资源和任务管理器。当用户提交一个应用程序时，需要提供一个用以跟踪和管理这个程序的ApplicationMaster，它负责向ResourceManager申请资源，并要求NodeManger启动可以占用一定资源的任务。由于不同的ApplicationMaster被分布到不同的节点上，因此它们之间不会相互影响。如图中所示，YARN主要是由Resource Manager、NodeManager和ApplicationMaster组成：

1）Resource Manager

RM是全局的资源管理器，负责整个系统的资源管理和分片，主要又两个组件组成：调度器和应用程序管理器

• 调度器：根据容量、队列等限制条件，将系统中的资源分配给各个正在运行的应用程序
• 应用程序管理器：负责管理整个系统中所有的应用程序，包括应用程序提交、与调度器协商资源以启动ApplicationMaster、监控ApplicationMaster运行状态并在失败时重启等

2）Application Master
用户提交的每个应用程序均包含1个AM，主要功能包括

• 与ResourceManager调度器协商以获取资源
• 将得到的任务进一步分配给内部的任务
• 与NodeManager通信以启动/停止任务
• 监控所有任务运行状态并在任务运行失败时重新为任务申请资源以重启任务等

3）NodeManager

NM是每个节点上的资源和任务管理器，它不时向RM汇报本节点上的资源使用情况和各个Container的运行状态，同时也会接收并处理来自AM的启动或停止请求。

4）Container

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

3.2 YARN工作流程

当用户向YARN提交一个应用程序后，YARN分两个阶段运行该应用程序：第一阶段是启动ApplicationMaster；第二阶段是由ApplicationMaster创建应用程序，为它申请资源并监控整个运行过程，直至运行完成。

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为任务设置好运行环境后，将任务启动命令写到一个脚本中，并通过运行该脚本启动任务。
7. 各个任务通过某个RPC协议向ApplicationMaster汇报自己的状态和进度，使ApplicationMaster能够随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。在应用程序运行过程中，用户可随时通过RPC向ApplicationMaster查询应用程序的当前运行状态。
8. 应用程序运行完成后，ApplicationMaster通过RPC协议向ResourceManager注销并关闭自己。

4、HDFS分布式文件系统

HDFS全称Hadoop Distribute File System，是分布式计算中数据存储管理的基础，解决了海量数据的存储问题，具有高容错性、可扩展性，适合大数据量处理和流式批处理，可运行于廉价的服务器上。传统的HDFS架构采用master/slave架构，一个HDFS集群是由一个Namenode和一定数目的Datanodes组成。

• NameNode是master，它是一个中心服务器，是这个集群的管理者，负责管理HDFS的命名空间（NameSpace）、配置副本策略和数据块（Block）映射信息，同时也会处理客户端读写请求。
• DataNode是slave，NameNode下达命令，DataNode执行实际的操作。集群中的DataNode负责管理所在节点上的存储，主要负责存储实际的数据块和执行数据块的读/写操作。
• Client负责与用户交互，同时可以执行以下操作
  • 文件切分，文件上传HDFS的时候，Client将文件切分成一个一个的Block，然后进行存储。
  • 与NameNode交互，获取文件的位置信息
  • 与DataNode交互，读取或者写入数据
  • Client提供一些命令来管理HDFS，比如启动或者关闭HDFS
  • Client可以通过一些命令来访问HDFS
• Secondary NameNode：并非NameNode的热备。当NameNode挂掉的时候，它并不能马上替换NameNode并提供服务
  • 辅助NameNode，分担其工作量。
  • 定期合并fsimage和fsedits，并推送给NameNode。
  • 在紧急情况下，可辅助恢复NameNode

备注：HDFS分布式文件系统有关知识在“分布式文件系统HDFS概述”中详细介绍了，这里只做简单的描述。

5、总结

本文主要介绍了分布式计算框架Hadoop的三大组件：MapReduce、YARN和HDFS。MapReduce是整个分布式计算的基础，完成分布式任务的数据拆分、归并排序和汇总工作；YARN是通用的资源调度框架，不仅仅局限于MapReduce，也支持Spark、Storm、Flume等在线和流式计算框架；HDFS作为分布式文件系统，支持大数据量的数据存储、高吞吐访问和横向扩展。

---

参考资料：文章来源地址https://www.toymoban.com/news/detail-455248.html

1. 《Hadoop技术内幕 深入理解MapReduce架构设计与实现原理》，董西成著
2. 《Hadoop技术内幕：深入解析YARN架构设计与实现原理》，董西成著
3. 分布式文件系统HDFS概述
4. https://blog.csdn.net/qq_41858402/article/details/108207777
5. https://blog.csdn.net/qq_40589204/article/details/118160989
6. https://hadoop.apache.org/docs/stable/hadoop-yarn/hadoop-yarn-site/YARN.html