MapReduce作业运行机制

1、mapreduce的体系结构

2、mapreduce作业运行机制

2.1运行图

2.2运行解析

2.2.1作业的提交

1)此方法调用submit(). 在Submit()方法里面连接JobTracker，即生成一个内部JobSummitter（实际上是new JobClient(),在new JobClient()里面生成一个JobSubmissionProtocol接口(JobTracker实现了此接口)对象jobSubmitClient(是它连接或对应着JobTracker)），在Submit()方法里面也调用JobClient.submitJobInternal(conf)方法返回一个RunningJob（步骤1）

2)参数true说明要调用方法jobClient.monitorAndPrintJob()即检查作业的运行情况(每秒一次)，如果有变化就报告给控制台
jobClient.submitJobInternal()所实现的提交作业过程如下：
3) 向Jobtracker请求一个新的job ID(步骤2)
4) 检查作业的输出路径，如果未指定或已存在则不提交作业并抛错误给程序；
5) 计算并生成作业的输入分片，如果路径不存在则不提交作业并抛错误给程序；
6) 将运行作业所需要的资源（包括作业jar文件，配置文件和计算所得的输入分片）复制到jobtracker的文件系统中以job id命名的目录下(即HDFS中)。作业jar副本较多(mapred.submit.replication = 10)(步骤3)
7) 告知jobtracker作业准备执行（真正的提交作业jobSubmitClient.submitJob()）(步骤4)

2.2.3作业的初始化

1)jobtracker接收到对其submitJob()方法的调用后，将其放入内部队列，交由job scheduler进行调度，并对其进行初始化，包括创建一个正在运行作业的对象---封装任务和记录信息(步骤5)
2)为了创建任务运行列表，job scheduler首先从共享文件系统中获取已计算好的输入分片信息(步骤6)，然后为每个分片创建一个map任务；
3)创建的reduce任务数量由Job的mapred.reduce.task属性决定(setNumReduceTasks()设置)，schedule创建相应数量的reduce任务。任务在此时被指定ID。
4)除了map和reduce任务，还有setupJob和cleanupJob需要建立：由tasktrackers在所有map开始前和所有reduce结束后分别执行，这两个方法在OutputCommitter中(默认是FileOutputCommitter)。setupJob()创建输出目录和任务的临时工作目录，cleanupJob()删除临时工作目录。

2.2.4作业的分配

1)每个tasktracker定期发送心跳给jobtracker，告知自己还活着，并附带消息说明自己是否已准备好接受新任务。jobtracker以此来分配任务，并使用心跳的返回值与tasktracker通信(步骤7)。Jobtracker利用调度算法先选择一个job然后再选此job的一个task分配给tasktracker.
2)每个tasktracker会有固定数量的map和reduce任务槽，数量有tasktracker核的数量和内存大小来决定。jobtracker会先将tasktracker的所有的map槽填满，然后才填此tasktracker的reduce任务槽。
3)Jobtracker分配map任务时会选取与输入分片最近的tasktracker，分配reduce任务用不着考虑数据本地化。

2.2.5任务的执行

用到上面提到的setupJob()
1) tasktracker分配到一个任务后，首先从HDFS中把作业的jar文件及运行所需要的全部文件(DistributedCache设置的)复制到tasktracker本地(步骤8)；
2) 接下来tasktracker为任务新建一个本地工作目录，并把jar文件的内容解压到这个文件夹下；
3) tasktracker新建一个taskRunner实例来运行该任务(步骤9)；
4) TaskRunner启动一个新的JVM来运行每个任务(步骤10)，以便客户的map/reduce不会影响tasktracker。

2.2.6进度和状态的更新

一个作业和它的每个任务都有一个状态，包括：作业或任务的运行状态(running, successful, failed)，map和reduce的进度，计数器值，状态消息或描述。

map进度标准是处理输入所占比例，reduce是copy\merge\reduce整个进度的比例。
Child JVM有独立的线程每隔3秒检查任务更新标志，如果有更新就会报告给此tasktracker；
tasktracker每隔5秒给jobtracker发心跳；
job tracker合并这些更新，产生一个表明所有运行作业及其任务状态的全局试图。
JobClient.monitorAndPrintJob()每秒查询这些信息。

2.2.7作业的完成

当jobtracker收到最后一个任务(this will be the special job cleanup task)的完成报告后，便把job状态设置为successful。
Job得到完成信息便从waitForCompletion()返回。
最后，jobtracker清空作业的工作状态，并指示tasktracker也清空作业的工作状态(如删除中间输出)。

2.3失败解析

2.3.1 任务失败

1)子任务失败。当map或者reduce子任务中的代码抛出异常，JVM进程会在退出之前向服进程tasktracker进程发送错误报告,tasktracker会将此（任务尝试）task attempt标记为failed状态，释放一个槽以便运行另外一个任务

2)jvm失败。JVM突然退出，即JVM错误，这时tasktracker会注意到进程已经退出，标记为failed

另外：

1)任务失败有重试机制，重试次数map任务设置是mapred.map.max.attempts属性控制，reduce是mapred.reduce.max.attempts属性控制。

2)一些job可以任务完成总体的一部分就能够接受，这个百分比由mapred.map.failures.precent和mapred.reduce.failures.precent参数控制。

3)任务尝试(task attempt)是可以中止(killed)的。

2.3.2 tasktracker失败

作业运行期间，tasktracker会通过心跳机制不断与系统jobtracker通信，如果某个tasktracker运行缓慢或者失败，出现故障。tasktracker就会停止或者很少想jobtracker发送心跳，jobtracker会注意到此tasktracker发送心跳的情况，从而将此tasktracker从等待任务调度的tasktracker池中移除，

1) 如果是map并且成功完成的话, jobtracker会安排此tasktracker上一成功运行的map任务返回

2) 如果是reduce并且成功的话，数据直接使用,因为reduce只要执行完了的就会把输出写到Hdfs上
3) 如果他们属于未完成的作业的话，reduce阶段无法获取改tasktracker上的本地map输出文件，任何任务都需要重新调度

另外，即使tasktracker没有失败，如果它上面的失败任务远远高于集群的平均失败任务数，也会被列入黑名单。可以通过重启从jobtracker的黑名单移除。

2.3.3 jobtracker失败

jobtracker失败应该说是最严重的一种失败方式了，而且在Hadoop中存在单点故障的情况下是相当严重的，因为在这种情况下作业最终失败，尽管这种故障的概率极小。未来版本可以通过启动多个jobtracker，在这种情况只运行一个主的jobtracker.通过一种机制来确定那个是主的jobtracker.

MapReduce作业运行机制相关推荐

经典MapReduce作业和Yarn上MapReduce作业运行机制
一.经典MapReduce的作业运行机制如下图是经典MapReduce作业的工作原理: 1.1 经典MapReduce作业的实体经典MapReduce作业运行过程包含的实体: 客户端,提交MapR ...
MapReduce的运行机制及共同好友
1 .MapReduce的运行机制详解 1.1:MapTask 工作机制整个Map阶段流程大体如上图所示. 简单概述:inputFile通过split被逻辑切分为多个split文件,通过Record ...
【大数据day14】——MapReduce的运行机制详解（案列：Reduce 端实现 JOIN， Map端实现 JOIN，求共同好友）
文章目录 1 .MapReduce的运行机制详解 1.1:MapTask 工作机制详细步骤配置 1.2 :ReduceTask 工作机制详细步骤 1.3:Shuffle 过程 2. 案例: Re ...
什么是MapReduce？MapReduce的运行机制是什么？MapReduce的实现过程
1. MAPREDUCE原理篇(1) Mapreduce是一个分布式运算程序的编程框架,是用户开发"基于hadoop的数据分析应用"的核心框架: Mapreduce核心功能是将用户 ...
如果遇到Hadoop集群正常，MapReduce作业运行出现错误，如何来查看作业运行日志（图文详解）...
这个时候我们可以进入logs下的userlogs 备注:userlogs目录下有很多个以往运行的作业,我选择最新的最大编号的作业,就是我们当前运行作业的日志.然后找到stderr stdout sys ...
MapReduce 运行机制
Hadoop 中的MapReduce是一个使用简单的软件框架,基于它写出来的应用程序能够运行在由上千个商用机器组成的大型集群上,并以一种可靠容错式并行处理TB级别的数据集. 一个MapReduce作业 ...
MapReduce运行机制
相关链接 MapReduce中Shuffle机制详解--Map端Shuffle链接 MapReduce中Shuffle机制详解--Reduce端Shuffle链接 MapReduce将作业job的 ...
MapReduce作业在Hadoop完全分布式集群上运行的问题与思考（持续更新）
1.集群已搭建好且通过了WordCount测试,但是在eclipse上开发的程序却仍然是只在namenode上运行不知道是不是没有配置好eclipse上的Map/Reduce Locations,个 ...
MapTask并行度决定机制、FileInputFormat切片机制、map并行度的经验之谈、ReduceTask并行度的决定、MAPREDUCE程序运行演示（来自学笔记）
1.3 MapTask并行度决定机制 maptask的并行度决定map阶段的任务处理并发度,进而影响到整个job的处理速度那么,mapTask并行实例是否越多越好呢?其并行度又是如何决定呢? 1.3 ...
MapReduce运行机制-Map阶段
MapTask 运行机制整个Map阶段流程大体如上图所示. 简单概述:inputFile通过split被逻辑切分为多个split文件,通过Record按行读取内容给map(用户自己实现的)进行处理, ...

MapReduce作业运行机制

MapReduce作业运行机制相关推荐

最新文章

热门文章