您现在的位置是:首页 > 文章详情

MapReduce深度分析(二)

日期:2016-04-20点击:466

MapReduce深度分析(二)

五、JobTracker分析

JobTracker是hadoop的重要的后台守护进程之一,主要的功能是管理任务调度、管理TaskTracker、监控作业执行、运行作业容错机制等。

首先启动interTrackerServer,将端口配置为mapred.job.tracker绑定的地址和端口。interTrackerServer提供两种用途:

接收和处理TaskTracker的heartbeat请求,必须实现InterTrackerProtocol接口及协议。

接收和处理JobClient请求,submitJob、killJob等,必须实现JobSubmissionProtocol接口及协议。

其次启动一个infoServer,运行StatusHttpServer,提供web服务。

最后,启动5个JobTracker子线程,各子线程功能如下:

ExpireLaunchingTasks:用于停止那些未在超时内报告进度的Task ExpireTrackers:用于停止那些可能已经当掉的TaskTracker RetireJobs:用于清除那些已经完成很长时间还存在队列里的作业 JobInitThread:用于初始化用户作业 TaskCommitQueue:用于调度Task的那些所有与FileSystem操作相关的处理,并记录Task的状态信息。

JobTracker的启动分析:

主要有两个函数startTracker和offerService()函数。

1、JobTracker.startTracker()

这个函数在启动时被调用主要有两个步骤:

步骤1、构造JobTracker对象result,启动两个RPC服务,并等待JobTracker退出安全模式,如果构造出错,睡眠1秒

步骤2、调用静态方法JobEndNotifier.startNotifier()创建一个线程,从延迟队列JobEndNotifier.queue中取出一个JobEndStatusInfo对象,然后通过sendNotification()调用httpNotification()构造一个HttpClient对象执行相应的http请求。

2、JobTracker.offerService()

负责创建启动5个重要的线程及其回收线程资源。

六、TaskTracker分析

TaskTracker的主要任务是执行JobTracker分发的任务。

TaskTracker启动分析:

TaskTracker.TaskTracker()-->TaskTracker.run()

askTracker.TaskTracker()用于构造TaskTracker类的对象,在上述代码中调用了TaskTracker(conf)含参构造函数,首先为TaskTracker设置JobTracker地址,然后配置启动一个StatusHttpServer对象,将服务绑定到作业配置项tasktracker.http.prot中指定的地址和端口,提供web服务,用于向用户提供web界面查询任务执行情况的服务。最后调用initialize()完成TaskTracker初始化。

TaskTracker.run()是TaskTracker的主线程核心函数,调用TaskTracker.offerService()链接JobTracker并开始提供服务。如果出现网络故障,需等待5s,然后重试,根据异常类型可能还需先调用TaskTracker.close完成清理,将所有任务置为失败,关闭RPC server,将TaskTracker.running置为false,清理本地的Map计算输出,关闭取Map计算结果的线程mapEventsFetcher等,清理完成后调用initialize()重新初始化。

TaskTracker核心子线程:

MapEventsFetcher线程、taskCleanup线程、TaskRunner线程。

七、心跳机制实现分析

在hadoop的master启动的时候会开启一个IPC Server以等待Slave的心跳数据包,Slave启动时会主动链接master,并周期性的每个3s向master发送一个心跳包。slave通过这个心跳包将自己的状态告诉master,然后master在通过心跳包的返回值向slave节点传送执行指令,整个过程就是hadoop的心跳检测机制。

八、作业创建分析

作业的创建主要由JobClient类负责完成。JobClient也是用户作业和JobTracker交互的重要接口,可以用来提交作业、跟踪作业的状态、访问子任务的报告、日志等,获取mapreduce集群状态信息等。

JobClient在创建作业时执行的主要操作包括:检查输入/输出的有效性,计算作业的Splits,复制作业的jar包和配置文件到HDFS的mapred系统目录,最后提交作业给JobTracker并跟踪作业执行状态。整个作业的创建都在JobClient.runJob()函数中执行。

创建流程:

步骤1、首先根据传入的JobConf参数构造JobClient对象,在JobClient的构造函数中会调用JobClient的init方法,通过其连接到JobTracker。

步骤2、调用JobClient.submitJob()向JobTracker提交作业,根据返回的RunningJob接口每隔1s检查一次作业的状态,如果执行完毕退出跟踪,如果出错,杀死该作业。

初始化:

步骤1、从配置文件读取mapred.job.tracker,判断是否为本地执行的任务,初始化JobClient的JobSubmitClient。

步骤2、如果是本地任务,调用LocalJobRunner来初始化。

步骤3、如果非本地任务,使用RPC机制来构造一个JobSubmissionProtocol接口的代理,即调用JobTracker.getAddress(conf)获得JobTracker地址,再通过JobClient参数调用createProxy方法初始化JobSubmitClient。

作业提交:

提交是通过在JobClient.runJob()函数中调用JobClient.submitJob()函数完成的。最终通过调用JobClient.submitJobInternal()函数提交到作业JobTracker,则该函数返回NetworkedJob的RunningJob对象用于跟踪作业。主要的执行流程如下:

步骤1、通过jobSubmitClient.getNewJobId()获取作业名。

步骤2、获得作业提交目录submitJobDit,并设置参数mapreduce.job.dir的值。

步骤3、获取Job的分布式缓存路径。

步骤4、获取作业配置文件目录,获取Reduce数目,以及本机Ip地址,并根据Reduce数目是否为零来检查输入/输出设置。 

步骤5、为用户作业创建Split输入分区。

步骤6、将JobConf的mapred.job.split.file项配置为job.split在HDFS上的绝对路径,根据splits[]数组大小设置Map任务数。

步骤7、得到作业队列名并设置ACL队列管理信息。

步骤8、将JobConf的内容写到HDFS的/${maprede.system.dir}/${jobid}/job.xml中。

步骤9、最后通过jobSubmitClient.submitJob()将名为JobId的作业提交给JobTracker。返回status变量是JobStatus对象,用于跟踪作业状态。

九、作业执行分析

在完成作业创建过程后再JobClient.submitJob()函数中通过JobSubmissionProtocol协议调用JobTracker.submitJob()函数提交作业到JobTracker,在submitJob()函数中主要负责用户的用户作业的初始化、构造JobInProgress对象,并初始化任务列表等,而任务的真正执行实际是由TaskTracker完成的。

JobTracker初始化

在submitJob()函数被调用之后,JobTracker就会接收到新的job请求,然后创建一个JobInProgress对象并通过它来来管理和调度任务。JobInProgress在创建的时候会初始化一系列与任务有关的参数。

1、JobTracker.submitJob()

submitJob()是JobTracker初始化的主函数。主要流程如下:

步骤1、得到用户组信息,返回作业状态,新建jobInfo。

步骤2、创建JobInProgress对象,不锁定JobTracker。

步骤3、核对队列是否处于running状态,并检查作业访问权限。

步骤4、检查作业是否因为无效的内存需求而不能运行。

步骤5、通过调用addJob(jobID,job)函数提交作业。该函数会返回JobStatus对象用于跟踪作业的运行状态。

2、JobInProgress.JobInProgress()

这个函数在JobTracker.submitJob()初始化作业中构造JobInProgress对象时调用执行。

步骤1、创建并初始化用于向JobClient汇报作业执行状态的JobStatus对象。

步骤2、设置用户名信息以及该JobInProgress对象的启动时间。

步骤3、在joobTrackerbTracker本地文件系统的${mapred.local.dir}/目录下创建${jobid}.jar,${jobid}.xml和${job.id}目录,并在该目录下创建job.xml文件。

步骤4、从作业配置JobConf中读取作业优先级、队列信息,Map任务数,Reduce任务数等。

步骤5、创建taskCompletionEvents列表。该列表用于JobTracker上跟踪该作业完成事件,初始化大小为Map任务数+Reduce任务数+10

步骤6、构建jobACLs,用于对用户作业进行ACL权限控制。

步骤7、设置Map、Reduce,以及每个Tracker上任务可以容忍失败的百分比。

步骤8、检查每个reduce的估计输入大小是否小于reduce大小的限制值,最后注册作业。

TaskTracker.startNewTask()

初始化完成之后,通过该函数启动一个新的任务。在心跳机制检测中,如果JobTracker返回了LaunchTaskAction指令,则TaskTracker在offerService中会调用TaskTracker.startNewTask()函数来处理新任务,即开始执行一个新的计算任务。

TaskTracker.localizeJob()

该函数在TaskTracker.startNewTask()中调用,主要负责计算任务初始化(本地化)并启动计算任务。

TaskRunner.run()

最终执行TaskRunner.run函数,TaskTracker最终调用TaskTracker.TaskInProgress.launchTask()创建计算线程来执行。

当神已无能为力,那便是魔渡众生
原文链接:https://yq.aliyun.com/articles/496771
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。

文章评论

共有0条评论来说两句吧...

文章二维码

扫描即可查看该文章

点击排行

推荐阅读

最新文章