hadoop中MapReduce和yarn的基本原理讲解

目录

1 MapReduce

1.1 执行流程

1.2 工作原理

Map任务处理

Reduce任务处理

MR的Shuffle过程

1.3  运行流程分析

2 yarn

2.1简介

2.2 组成

1. ResourceManager

2. NodeManager

3.ApplicationMaster(AM)

4. Container

2.3 工作原理


1 MapReduce

MapReduce是一种编程模型,用于大规模数据集(大于1TB)的并行运算。概念"Map(映射)"和"Reduce(归约)",和它们的主要思想,都是从函数式编程语言里借来的,还有从矢量编程语言里借来的特性。它极大地方便了编程人员在不会分布式并行编程的情况下,将自己的程序运行在分布式系统上。 当前的软件实现是指定一个Map(映射)函数,用来把一组键值对映射成一组新的键值对,指定并发的Reduce(归约)函数,用来保证所有映射的键值对中的每一个共享相同的键组。

记住这2点:

      1、MapReduce是一种分布式计算模型,是Google提出的,主要用于搜索领域,解决海量数据的计算问题。

      2、MR有两个阶段组成:Map和Reduce,用户只需实现map()和reduce()两个函数,即可实现分布式计算。

1.1 执行流程

hadoop中MapReduce和yarn的基本原理讲解

由图我们可以看到, MapReduce存在以下4个独立的实体。
1. JobClient:运行于client node,负责将MapReduce程序打成Jar包存储到HDFS,并把Jar包的路径提交到Jobtracker,由Jobtracker进行任务的分配和监控。
2. JobTracker:运行于name node,负责接收JobClient提交的Job,调度Job的每一个子task运行于TaskTracker上,并监控它们,如果发现有失败的task就重新运行它。
3. TaskTracker:运行于data node,负责主动与JobTracker通信,接收作业,并直接执行每一个任务。
4. HDFS:用来与其它实体间共享作业文件。

 

各实体间通过以下过程完成一次MapReduce作业。

  1. JobClient通过RPC协议向JobTracker请求一个新应用的ID,用于MapReduce作业的ID
  2. JobTracker检查作业的输出说明。例如,如果没有指定输出目录或目录已存在,作业就不提交,错误抛回给JobClient,否则,返回新的作业ID给JobClient
  3. JobClient将作业所需的资源(包括作业JAR文件、配置文件和计算所得得输入分片)复制到以作业ID命名的HDFS文件夹中
  4. JobClient通过submitApplication()提交作业
  5. JobTracker收到调用它的submitApplication()消息后,进行任务初始化
  6. JobTracker读取HDFS上的要处理的文件,开始计算输入分片,每一个分片对应一个TaskTracker
  7. askTracker通过心跳机制领取任务(任务的描述信息)
  8. TaskTracker读取HDFS上的作业资源(JAR包、配置文件等)
  9. TaskTracker启动一个java child子进程,用来执行具体的任务(MapperTask或ReducerTask)

  10. TaskTracker将Reduce结果写入到HDFS当中

1.2 工作原理

hadoop中MapReduce和yarn的基本原理讲解

Map任务处理

  1. 读取HDFS中的文件。每一行解析成一个<k,v>。每一个键值对调用一次map函数
  2. 重写map(),对第一步产生的<k,v>进行处理,转换为新的<k,v>输出
  3. 对输出的key、value进行分区
  4. 对不同分区的数据,按照key进行排序、分组。相同key的value放到一个集合中
  5. (可选) 对分组后的数据进行归约

Reduce任务处理

  1. 多个map任务的输出,按照不同的分区,通过网络复制到不同的reduce节点上
  2. 对多个map的输出进行合并、排序。
  3. 重写reduce函数实现自己的逻辑,对输入的key、value处理,转换成新的key、value输出
  4. 把reduce的输出保存到文件中

hadoop中MapReduce和yarn的基本原理讲解

MR的Shuffle过程

MapReduce计算模型主要由三个阶段构成:Map、Shuffle、Reduce。Map是映射,负责数据的过滤分类,将原始数据转化为键值对;Reduce是合并,将具有相同key值的value进行处理后再输出新的键值对作为最终结果;为了让Reduce可以并行处理Map的结果,必须对Map的输出进行一定的排序与分割,然后再交给对应的Reduce,这个过程就是Shuffle。Shuffle过程包含Map Shuffle和Reduce Shuffle。

1)Map Shuffle

在Map端的shuffle过程就是对Map的结果进行分区、排序、分割,然后将属于同一个分区的输出合并在一起并写在磁盘上,最终得到一个分区有序的文件。分区有序的含义是Map输出的键值对按分区进行排列,具有相同partition值的键值对存储在一起,每个分区里面的键值对又按key值进行升序排序(默认),大致流程如下:

hadoop中MapReduce和yarn的基本原理讲解

2)Reduce Shuffle

Reduce任务通过HTTP向各个Map任务拖取它所需要的数据。Map任务成功完成后,会通知父TaskTracker状态已经更新,TaskTracker进而通知JobTracker(这些通知在心跳机制中进行)。所以,对于指定作业来说,JobTracker能记录Map输出和TaskTracker的映射关系。Reduce会定期向JobTracker获取Map的输出位置,一旦拿到输出位置,Reduce任务就会从此输出对应的TaskTracker上复制输出到本地,而不会等到所有的Map任务结束。

Copy过来的数据会先放入内存缓冲区中,如果内存缓冲区中能放得下这次数据的话就直接把数据写到内存中,即内存到内存merge。Reduce要向每个Map去拖取数据,在内存中每个Map对应一块数据,当内存缓存区中存储的Map数据占用空间达到一定程度的时候,开始启动内存中merge,把内存中的数据merge输出到磁盘上一个文件中,即内存到磁盘merge。在将buffer中多个map输出合并写入磁盘之前,如果设置了Combiner,则会化简压缩合并的map输出。Reduce的内存缓冲区可通过mapred.job.shuffle.input.buffer.percent配置,默认是JVM的heap size的70%。内存到磁盘merge的启动门限可以通过mapred.job.shuffle.merge.percent配置,默认是66%。

当属于该reducer的map输出全部拷贝完成,则会在reducer上生成多个文件(如果拖取的所有map数据总量都没有内存缓冲区,则数据就只存在于内存中),这时开始执行合并操作,即磁盘到磁盘merge,Map的输出数据已经是有序的,Merge进行一次合并排序,所谓Reduce端的sort过程就是这个合并的过程。一般Reduce是一边copy一边sort,即copy和sort两个阶段是重叠而不是完全分开的。最终Reduce shuffle过程会输出一个整体有序的数据块。

1.3  运行流程分析

hadoop中MapReduce和yarn的基本原理讲解

流程分析:

Map端:

1.每个输入分片会让一个map任务来处理,默认情况下,以HDFS的一个块的大小(默认为64M)为一个分片,当然我们也可以设置块的大小。map输出的结果会暂且放在一个环形内存缓冲区中(该缓冲区的大小默认为100M,由io.sort.mb属性控制),当该缓冲区快要溢出时(默认为缓冲区大小的80%,由io.sort.spill.percent属性控制),会在本地文件系统中创建一个溢出文件,将该缓冲区中的数据写入这个文件。

2.在写入磁盘之前,线程首先根据reduce任务的数目将数据划分为相同数目的分区,也就是一个reduce任务对应一个分区的数据。这样做是为了避免有些reduce任务分配到大量数据,而有些reduce任务却分到很少数据,甚至没有分到数据的尴尬局面。其实分区就是对数据进行hash的过程。然后对每个分区中的数据进行排序,如果此时设置了Combiner,将排序后的结果进行Combia操作,这样做的目的是让尽可能少的数据写入到磁盘。

3.当map任务输出最后一个记录时,可能会有很多的溢出文件,这时需要将这些文件合并。合并的过程中会不断地进行排序和combia操作,目的有两个:1.尽量减少每次写入磁盘的数据量;2.尽量减少下一复制阶段网络传输的数据量。最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量,这里可以将数据压缩,只要将mapred.compress.map.out设置为true就可以了。

4.将分区中的数据拷贝给相对应的reduce任务。有人可能会问:分区中的数据怎么知道它对应的reduce是哪个呢?其实map任务一直和其父TaskTracker保持联系,而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整个集群中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就ok了哦。

到这里,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函数

2 yarn

2.1简介

Yarn是Hadoop集群的资源管理系统。Hadoop2.0对MapReduce框架做了彻底的设计重构,我们称Hadoop2.0中的MapReduce为MRv2或者Yarn。 它的目标是将这两部分功能分开,也就是分别用两个进程来管理这两个任务:

 Yarn/MRv2最基本的想法是将原JobTracker主要的资源管理和job调度/监视功能分开作为两个单独的守护进程。有一个全局的ResourceManager(RM)和每个Application有一个ApplicationMaster(AM),Application相当于map-reduce job或者DAG jobs。ResourceManager和NodeManager(NM)组成了基本的数据计算框架。ResourceManager协调集群的资源利用,任何client或者运行着的applicatitonMaster想要运行job或者task都得向RM申请一定的资源。ApplicatonMaster是一个框架特殊的库,对于MapReduce框架而言有它自己的AM实现,用户也可以实现自己的AM,在运行的时候,AM会与NM一起来启动和监视tasks。

核心思想:将MP1中JobTracker的资源管理和作业调度两个功能分开,分别由ResourceManager和ApplicationMaster进程来实现。

  1)ResourceManager:负责整个集群的资源管理和调度。

  2)ApplicationMaster:负责应用程序相关的事务,比如任务调度、任务监控和容错等。

 YARN的出现,使得多个计算框架可以运行在一个集群当中。

  1)每个应用程序对应一个ApplicationMaster。

  2)目前可以支持多种计算框架运行在YARN上面比如MapReduce、Storm、Spark、Flink等。

2.2 组成

hadoop中MapReduce和yarn的基本原理讲解

从 YARN 的架构图来看,它主要由ResourceManager、NodeManager、ApplicationMaster和Container等以下几个组件构成。

ResourceManager:一个Cluster只有一个,负责资源调度、资源分配等工作。

NodeManager:运行在DataNode节点,负责启动Application和对资源的管理。

JobHistoryServer:负责查询job运行进度及元数据管理。

Containers:Container通过ResourceManager分配。包括容器的CPU、内存等资源。

Application Master:ResourceManager将任务给Application Master,然后Application Master再将任务给NodeManager。每个Application只有一个Application Master,运行在Node Manager节点,Application Master是由ResourceManager指派的。

job:是需要执行的一个工作单元:它包括输入数据、MapReduce程序和配置信息。job也可以叫作Application。

task:一个具体做Mapper或Reducer的独立的工作单元。task运行在NodeManager的Container中。

Client:一个提交给ResourceManager的一个Application程序。

1. ResourceManager

RM 是一个全局的资源管理器,负责整个系统的资源管理和分配。它主要由两个组件构成:

调度器(Scheduler)和应用程序管理器(Applications Manager,ASM)。

YARN 分层结构的本质是 ResourceManager。这个实体控制整个集群并管理应用程序向基础计算资源的分配。ResourceManager 将各个资源部分(计算、内存、带宽等)精心安排

给基础 NodeManager(YARN 的每节点代理)。ResourceManager 还与 ApplicationMaster 一起分配资源,与 NodeManager 一起启动和监视它们的基础应用程序。在此上下文中,ApplicationMaster 承担了以前的 TaskTracker 的一些角色, ResourceManager 承担了 JobTracker 的角色。

(1)调度器

调度器根据容量、队列等限制条件(如每个队列分配一定的资源,最多执行一定数量的作业等),将系统中的资源分配给各个正在运行的应用程序。该调度器是一个“纯调度器”,它不再从事任何与具体应用程序相关的工作。

(2)应用程序管理器

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

2. NodeManager

NodeManager管理一个YARN集群中的每个节点。NodeManager提供针对集群中每个节点的服务,从监督对一个容器的终生管理到监视资源和跟踪节点健康。MRv1通过插槽管理Map和Reduce任务的执行,而NodeManager 管理抽象容器,这些容器代表着可供一个特定应用程序使用的针对每个节点的资源。YARN继续使用HDFS层。它的主要 NameNode用于元数据服务,而DataNode用于分散在一个集群中的复制存储服务。

  1)单个节点上的资源管理;

  2)处理来自ResourceManager上的命令;

  3)处理来自ApplicationMaster上的命令。

3.ApplicationMaster(AM)

 ApplicationMaster管理一个在YARN内运行的应用程序的每个实例。ApplicationMaster 负责协调来自 ResourceManager 的资源,并通过 NodeManager 监视容器的执行和资源使用(CPU、内存等的资源分配)。请注意,尽管目前的资源更加传统(CPU 核心、内存),但未来会带来基于手头任务的新资源类型(比如图形处理单元或专用处理设备)。从 YARN 角度讲,ApplicationMaster 是用户代码,因此存在潜在的安全问题。YARN 假设 ApplicationMaster 存在错误或者甚至是恶意的,因此将它们当作无特权的代码对待。

  1)负责数据的切分;

  2)为应用程序申请资源并分配给内部的任务;

  3)任务的监控与容错。

4. Container

      对任务运行环境进行抽象,封装CPU、内存等多维度的资源以及环境变量、启动命令等任务运行相关的信息。比如内存、CPU、磁盘、网络等,当AM向RM申请资源时,RM为AM返回的资源便是用Container表示的。YARN会为每个任务分配一个Container,且该任务只能使用该Container中描述的资源。

  要使用一个YARN集群,首先需要来自包含一个应用程序的客户的请求。ResourceManager 协商一个容器的必要资源,启动一个ApplicationMaster 来表示已提交的应用程序。通过使用一个资源请求协议,ApplicationMaster协商每个节点上供应用程序使用的资源容器。执行应用程序时,ApplicationMaster 监视容器直到完成。当应用程序完成时,ApplicationMaster 从 ResourceManager 注销其容器,执行周期就完成了。

2.3 工作原理

 

hadoop中MapReduce和yarn的基本原理讲解

客户端程序向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

YARN工作原理简述

hadoop中MapReduce和yarn的基本原理讲解

 

Client提交作业到 YARN 上;

Resource Manager选择一个Node Manager,启动一个Container并运行Application Master实例;

Application Master根据实际需要向Resource Manager请求更多的Container资源(如果作业很小, 应用管理器会选择在其自己的 JVM 中运行任务);

Application Master通过获取到的Container资源执行分布式计算。

YARN工作原理详述

hadoop中MapReduce和yarn的基本原理讲解

1. 作业提交

client 调用 job.waitForCompletion 方法,向整个集群提交 MapReduce 作业 (第 1 步) 。新的作业 ID(应用 ID) 由资源管理器分配 (第 2 步)。作业的 client 核实作业的输出, 计算输入的 split, 将作业的资源 (包括 Jar 包,配置文件, split 信息) 拷贝给 HDFS(第 3 步)。最后, 通过调用资源管理器的 submitApplication() 来提交作业 (第 4 步)。

2. 作业初始化

当资源管理器收到 submitApplciation() 的请求时, 就将该请求发给调度器(scheduler), 调度器分配 container, 然后资源管理器在该 container 内启动应用管理器进程, 由节点管理器监控 (第 5 步)。MapReduce 作业的应用管理器是一个主类为 MRAppMaster 的 Java 应用,其通过创造一些 bookkeeping 对象来监控作业的进度, 得到任务的进度和完成报告 (第 6 步)。然后其通过分布式文件系统得到由客户端计算好的输入 split(第 7 步),然后为每个输入 split 创建一个 map 任务, 根据 mapreduce.job.reduces 创建 reduce 任务对象。

3. 任务分配

如果作业很小, 应用管理器会选择在其自己的 JVM 中运行任务。如果不是小作业, 那么应用管理器向资源管理器请求 container 来运行所有的 map 和 reduce 任务 (第 8 步)。这些请求是通过心跳来传输的, 包括每个 map 任务的数据位置,比如存放输入 split 的主机名和机架 (rack),调度器利用这些信息来调度任务,尽量将任务分配给存储数据的节点, 或者分配给和存放输入 split 的节点相同机架的节点。

4. 任务运行

当一个任务由资源管理器的调度器分配给一个 container 后,应用管理器通过联系节点管理器来启动 container(第 9 步)。任务由一个主类为 YarnChild 的 Java 应用执行, 在运行任务之前首先本地化任务需要的资源,比如作业配置,JAR 文件, 以及分布式缓存的所有文件 (第 10 步。最后, 运行 map 或 reduce 任务 (第 11 步)。YarnChild 运行在一个专用的 JVM 中, 但是 YARN 不支持 JVM 重用。

5. 进度和状态更新

YARN 中的任务将其进度和状态 (包括 counter) 返回给应用管理器, 客户端每秒 (通 mapreduce.client.progressmonitor.pollinterval 设置) 向应用管理器请求进度更新, 展示给用户。

6. 作业完成

除了向应用管理器请求作业进度外, 客户端每 5 分钟都会通过调用 waitForCompletion() 来检查作业是否完成,时间间隔可以通过 mapreduce.client.completion.pollinterval 来设置。作业完成之后, 应用管理器和 container 会清理工作状态, OutputCommiter 的作业清理方法也会被调用。作业的信息会被作业历史服务器存储以备之后用户核查。