大数据人官方⑤群

大数据人|中国大数据第一社区

 找回密码
 注册会员

扫一扫,访问微社区

Hadoop工作机制

2016-6-18 19:50| 发布者: admin| 查看: 21417| 评论: 0|原作者: Heaven Wang |来自: CSDN

摘要: 可以只用一行代码来运行MapReduce作业:JobClient.runJon(conf),Job作业运行时参与的四个实体: 1.JobClient 写代码,配置作业,提交作业。 2.JobTracker:初始化作业,分配作业,协调作业运行。这是 ...

可以只用一行代码来运行MapReduce作业:JobClient.runJon(conf),Job作业运行时参与的四个实体:


     1.JobClient 写代码,配置作业,提交作业。

     2.JobTracker:初始化作业,分配作业,协调作业运行。这是一个java程序,主类是JobTracker。

     3.TaskTracker:运行作业划分后的任务,即分配数据分配上执行Map或Reduce任务。

     4.HDFS:保存作业数据、配置信息等,保存作业结果。


Map/Reduce 作业总体执行流程:


     代码编写 ----> 作业配置  ---->  作业提交 ----> Map任务分配和执行 ----> 处理中间结果 ---->  Reduce任务分配与执行 ---->  输出结果


而对于每个作业的执行,又包含:

     输入准备 ----> 任务执行 ----> 输出结果


作业提交JobClient:


     JobClient的runJob方法产生一个Jobclient实例并调用其submitJob方法,然后runJob开始循环吗,并在循环中调用getTaskCompetionEvents方法,获得TaskCompletionEvent实例,每秒轮询作业进度(后面有介绍进度和状态更新),把进度写到控制台,作业完成后显示作业计数器,若失败,则把错误记录到控制台。


     submitJob方法作业提交的过程:


     1.向JobTracker请求一个新的JobId。

     2.检查作业相关路径,如果路径不正确就会返回错误。

     3.计算作业输入分片及其划分信息。

     4.将作业运行需要的资源(jar文件、配置文件等)复制到Shared HDFS,并

复制多个副本(参数控制,默认值为10)供tasktracker访问,也会将计算的分片复制到HDFS。

     5.调用JobTracker对象的submitJob()方法来真正提交作业,告诉JobTracker作业准备执行。


作业的初始化JobTracker:


     JobTracker收到submitJob方法调用后,会把调用放入到一个内部队列,由作业调度器(Job scheduler)进行调度并对其初始化。Job初始化即创建一个作业对象。


     当作业被调度后,JobTracker会创建一个代表这个作业的JobInProgress对象,并将任务和记录信息封装在这个对象中,以便跟踪任务状态和进程。


     初始化过程就是JobInProgress对象的initTasks方法进行初始化的。


     初始化步骤:


          1.从HDFS中读取作业对应的job.split信息,为后面的初始化做好准备。

          2.创建并初始化map和reduce任务。根据数据分片信息中的个数确定map task的个数,然后为每个map task生成一个TaskInProgress对象来处理数据分片,先将其放入nonRunningMapCache,以便JobTracker分配任务的时候使用。接下来根据JobConf中的mapred.reduce.tasks属性利用setNumReduceTasks()方法设置reduce task的数量,然后同map task创建方式。

          3.最后就是创建两个初始化task,进行map和reduce的初始化。


任务的分配JobTracker:


    消息传递HeartBeat: tasktracker运行一个简单循环定期发送心跳(heartbeat)给JobTracker。由心跳告知JobTracker自己是否存活,同时作为消息通道传递其它信息(请求新task)。作为心跳的一部分,tasktracker会指明自己是否已准备好运行新的任务,如果是,jobtracker会分配它一个任务。


    分配任务所属于的作业:在Jobtracker分配任务前需先确定任务所在的作业。后面会介绍到各种作业调度算法,默认是一个FIFO的作业调度。


    分配Map和Reduce任务:tasktracker有固定数量的任务槽,一个tasktracker可以同时运行多个Map和Reduce任务,但其准确的数量由tasktracker的核的数量和内存大小决定。默认调度器会先填满Map任务槽,再填Reduce任务槽。jobtracker会选择距离离分片文件最近的tasktracker,最理想情况下,任务是数据本地化(data-local)的,当然也可以是机架本地化(rack-local),如果不是本地化的,那么他们就需要从其他机架上检索数据。Reduce任务分配很简单,jobtracker会简单的从待运行的reduce任务列表中选取下一个来执行,不用考虑数据本地化。


任务的执行TaskTracker:


     TaskTracker收到新任务后,就要在本地运行任务了,运行任务的第一步就是通过localizedJob将任务本地化所需要的注入配置、数据、程序等信息进行本地化。


     1.本地化数据:从共享文件系统将job.split 、job.jar (在分布式缓存中)复制本地,将job配置信息写入job.xml。

     2.新建本地工作目录:tasktracker会加压job.jar文件到本工作目录。

     3.调用launchTaskForJob方法发布任务(其中会新建TaskRunner实例运行任务),如果是Map任务就启用MapTaskRunner,对于Reduce就是ReduceTaskRunner。

   

  在这之后,TaskRunner会启用一个新的JVM来运行每个Map/Reduce任务,防止程序原因而导致tasktracker崩溃,但不同任务间重用JVM还是可以的,后续会讲到任务JVM重用


     对于单个Map,任务执行的简单流程是:


     1.分配任务执行参数

     2.在Child临时文件中添加map任务信息(Child是运行Map和Reduce任务的主进程)

     3.配置log文件夹,配置map任务的通信和输出参数

     4.读取input split,生成RecordReader读取数据

     5.为Map生成MapRunnable,依次从RecordReader中接收数据,并调用Map函数进行处理。

     6.最后将map函数的输出调用collect收集到MapOutputBuffer(参数控制其大小)中。


Streaming和Pipes:

     

     Streaming和Pipes都运行特殊的Map和Reduce任务,目的是运行用户提供的可执行程序并与之通信。


     Streaming:使用标准输入输出Streaming与进程进行通信。


     Pipes:用来监听套接字,会发送一个端口号给C++程序,两者便可建立链接。

     

进度和状态更新:


     一个作业和它的任务都有状态(status),其中包括:运行成功失败状态、Map/Reduce进度、作业计数器值、状态消息。


     状态消息与客户端的通信:


     1.对于Map任务Progress的追踪:progress是已经处理完的输入所占的比例。

     2.对于Reduce:稍复杂,reduce任务分三个阶段(每个阶段占1/3),复制、排序和Reduce处理,若reduce已执行一半的输入的话,那么任务进度便是1/3+1/3+1/6=5/6。

     3.任务计数器:任务有一组计数器,负责对任务运行各个事件进行计数。

     4.任务进度报告:如果任务报告了进度,便会设置一个标记以表明状态将被发送到tasktracker。有一个独立线程每隔三秒检查一次此标记,如果已设置,则告知tasktracker当前状态。

     5.tasktracker进度报告:tasktracker会每隔5秒(这个心跳是由集群大小决定,集群越大时间会越长)发送heartbeat到jobtracker,并且tasktracker运行的所有状态都会在调用中被发送到jobtracker。

     6.jobtracker合并各任务报告:产生一个表明所有运行作业机器所含任务状态的全局视图。

     前面提到的JobClient就是通过每秒查询JobTracker来接收最新状态,而且客户端JobClient的getJob方法可以得到一个RunningJob的实例,其包含了作业的所以状态信息。

     

作业的完成:


     当jobtracker收到作业最后一个任务已完成的通知后,便把作业状态设置成成功。JobClient查询状态时,便知道任务已成功完成,于是JobClient打印一条消息告知用户,然后从runJob方法返回。


     如果jobtracker有相应设置,也会发送一个Http作业通知给客户端,希望收到回调指令的客户端可以通过job.end.notification.url属性来进行设置。


     jobtracker情况作业的工作状态,指示tasktracker也清空作业的工作状态,如删除中间输出。

     

失败

     

     实际情况下,用户的代码存在软件错误进程会崩溃,机器也会产生故障,但Hadoop能很好的应对这些故障并完成作业。


     1.任务失败    

     子任务异常:如Map/Reduce任务中的用户代码抛出异常,子任务JVM进程会在退出前向父进程tasktracker发送错误报告,错误被记录用户日志。tasktracker会将此次task attempt标记为tailed,并释放这个任务槽运行另外一个任务。


     子进程JVM突然退出:可能由于JVM bug导致用户代码造成的某些特殊原因导致JVM退出,这种情况下,tasktracker会注意到进程已经退出,并将此次尝试标记为failed。


     任务挂起:一旦tasktracker注意一段时间没有收到进度更新,便会将任务标记为failed,JVM子进程将被自动杀死。任务失败间隔时间通常为10分钟,可以以作业或者集群为基础设置过期时间,参数为mapred.task.timeout。注意:如果参数值设置为0,则挂起的任务永远不会释放掉它的任务槽,随着时间的推移会降低整个集群的效率。


     任务失败尝试次数:jobtracker得知一个tasktracker失败后,它会重新调度该任务执行,当然,jobtracker会尝试避免重新调度失败过的tasktracker任务。如果一个任务尝试次数超过4次,它将不再被重试。这个值是可以设置的,对于Map任务,参数是mapred.map.max.attempts,对于reduce任务,则由mapred.reduce.max.attempts属性控制。如果次数超过限制,整个作业都会失败。当然,有时我们不希望少数几个任务失败就终止运行的整个作业,因为即使有些任务失败,作业的一些结果可能还是有用的,这种情况下,可以为作业设置在不触发作业失败情况下的允许任务失败的最大百分比,Map任务和Reduce任务可以独立控制,参数为mapred.max.map.failures.percent 和mapred.max.reduce.failures.percent。


     任务尝试中止(kill):任务终止和任务失败不同,task attempt可以中止是因为他是一个推测副本或因为它所处的tasktracker失败,导致jobtracker将它上面的所有task attempt标记为killed。被终止的task attempt不会被计入任务运行尝试次数,因为尝试中止并不是任务的错。


     2.tasktracker失败

     tasktracker由于崩溃或者运行过慢而失败,他将停止向jobtracker发送心跳(或很少发送心跳)。jobtracker注意已停止发送心跳的tasktracker(过期时间由参数mapred.tasktracker.expiry.interval设置,单位毫秒),并将它从等待调度的tasktracker池中移除。如果是未完成的作业,jobtracker会安排次tasktracker上已经运行成功的Map任务重新运行,因为此时reduce任务已无法访问(中间输出存放在失败的tasktracker的本地文件系统上)。


     即使tasktracker没有失败,也有可能被jobtracker列入黑名单。如果tasktracker上面的失败任务数量远远高于集群的平均失败任务次数,他就会被列入黑名单,被列入黑名单的tasktracker可以通过重启从jobtracker黑名单中移除。


     3.jobtracker失败

     老版本的JobTracker失败属于单点故障,这种情况下作业注定失败。

分享到: QQ空间 新浪微博 腾讯微博 人人网 飞信 百度搜藏


鲜花

握手

雷人

路过

鸡蛋

最新评论

关闭

站长推荐上一条 /2 下一条


id="mn_portal" >首页Portalid="mn_P18" onmouseover="navShow('P18')">应用id="mn_P15" onmouseover="navShow('P15')">技术id="mn_P37" onmouseover="showMenu({'ctrlid':this.id,'ctrlclass':'hover','duration':2})">前沿id="mn_P36" onmouseover="navShow('P36')">宝箱id="mn_P61" onmouseover="showMenu({'ctrlid':this.id,'ctrlclass':'hover','duration':2})">专栏id="mn_P65" >企业id="mn_forum_2" >社区BBSid="mn_Nd633" >导航id="mn_Nbe89" >通讯录id="mn_N4033" >赞+1 折叠导航 关注微信 关注微博 关注我们

QQ|广告服务|关于我们|鄂ICP备14012176号-2|Archiver|手机版|小黑屋|大数据人    

GMT+8, 2017-6-23 16:36 , Processed in 0.497535 second(s), 23 queries .

Powered by 小雄! X3.2

© 2014-2016 bigdataer Inc.

返回顶部