skyFS-mapreduce0.5版作业调度分析

skyFS-mapreduce 设计文档-Mapreduce Job Scheduler

Staff Date Content version

sai 12/11/12 Create 0.5

1.概要

　　本文主要论述skyFS-mapreduce下面mapreduce部分job调度的问题。简单介绍了mapreduce部分job正常调度时是如何分配，何时分配及何时执行的,比如group_mapper何时分配mapper及执行mapper，是一次性分配好排队等待执行；还是动态分配，每个mapper执行完再根据snode空闲情况进行分配。由于本人经验有限以及工作疏忽，难免有分析不周或错误的地方，如有发现，望请指正，不甚感激！

　　

1.1 背景

　　了解mapreduce的Job调度问题，可以知晓Job是在何时分配，初始化，执行，分配子job，以及分配snode或者task，可以发现mapreduce框架的不足之处。也为将来改进mapreduce框架，特别是Job调度算法提供参考依据。

　　

1.2 相关模块

　　

　　mr_pids_manager

　　client

　　jobs_manager

　　mr_jobs_sup

　　mr_tasks_sup

　　mr_uuids

　　mr_resource_manager

　　job_master

　　mr_pids_manager

　　mr_snode_task_manager

　　mr_snode_task

　　mr_group_mapper

　　mr_mapper

　　

Job调度

2.1系统初始化

　　mr_pids_manager:start_link(),

　　application:start(client),

　　mr_snode_task_manager:start_link(),

　　jobs_manager:start_link(),

　　mr_jobs_sup:start_link(),

　　mr_tasks_sup:start_link(),

　　mr_uuids:start_link(),

　　mr_resource_manager:start_link()

　　

　　

　　

2.2 启动Job

　　{ok,JobId0} = jobs_manager:start_job(Type, Args,self())

　　

2.3 Job启动步骤

　　

2.3.1首先，jobs_manager，通过uuid生成一个新的JobID，然后在启动start_job_master并返回master的PID，最后启动job_master,并记录到ets。

业务主要在start_job_master中完成。



2.3.2 现在分析start_job_master，首先会调用mr_resource_manager定位master Node，然后通过RPC远程调用在master Node上面启动Job。



2.3.3 mr_snode_task_manager会先启动job监督树，然后通过监督树来启动job，

启动时会先初始化job。



2.3.4 job_master启动后先通过mr_util和mr_pids_manager向ets注册



2.3.5 job_master先确认master Node是可连接的，然后启动mr_snode_task_manager.



2.3.6 mr_snode_task_manager也先启动监督树，然后再通过mr_snode_task启动具体job task任务



2.3.7 mr_snode_task启动后先初始化，加载job task配置信息，根据配置信息，

调用具体的mr_group_mapper来执行mapreduce任务，然后通知job_master任务已经开始，让job_master来处理mapreduce相关任务，比如map处理中，reduce处理中，超时重连等。



2.3.8 mr_group_mapper启动后先初始化，然后向mr_snode_task汇报进度信息。

之后调用mr_group_mapper中的mapper(run)来具体执行。在group mapper会根据配置信息，将子job信息初始化，并分配给job_master来生成子job。分配时，会根据现有集群中snode资源信息及子job类型来指定子job执行的最优snode，但是分配的snode并不一定立即执行子job。另外，如果子job,还是嵌套的任务，则在执行时继续拆分再分配其子job执行。



源码解析

接下来对部分关键代码做个简单的剖析

3.1首先来看一下mr_group_mapper中run部分。我们可以看到其通过语句

SubJobs = start_link_sub_job(Id, SubJobType, KeyValuePairs)

生成了子job列表，然后在通过lists：foreach/2和job_master:start_job(Mpid)函数来启动每个子job。







3.2 接下来看start_link_sub_job是如何分配子job的。我们可以看到会在本地job_master通过job_master:start_link/5创建一个进程Mpid，并将该Mpid保存到子job信息中。



3.3最后，我们来看下job_master:start_job/1是如何启动子job的。先组织好Task信息，我们可以看到node是undefined，即未分配的，然后调用start_running_task/1

来执行组织好的Task任务。







3.4 接着我们再看start_running_task/1里面是如何启动子job的。我们可以看到它先为子job分配TaskID，然后获得MasterNode，再通过MasterNode来调用mr_resource_manager模块来分配子job执行的结点。如果分配成功，子job就立即执行，否则等待一段时间（默认值：3000ms）后发送request_node_timeout消息。



3.5现在我们就需要看下未分配结点或者分配结点失败后是如何处理的。上代码：



很简单重新执行start_running_task/1。也就是说未分配成功的等候一段时间后继续等待资源分配，直到成功分配到资源执行。

结论

通过上面的分析，大致了解了skyFS-mapreduce中mr_group_mapper中是如何分配子job的。我们可以知道目前是根据job类型及集群中snode资源，分配最优snode来执行该job，但是job分配后并不一定立即执行，需要等分配的snode空闲，然后在从分配给该snode的一个或多个job中选取一个或多个来执行，执行的时候才加载job的配置信息进行初始化，而不是在分配时就已经初始化。

表面来看是一次分配，然后在各个snode执行（并行或串行）的。但是实际上虽然一次分配了，但是没有立即执行，而是需要先看集群资源时候足够。这种框架能胜任mapreduce任务，任务分配的也比较均衡，也相对充分的利用了集群资源。

　　目前的mapreducejob调度算法是是被动式分配的，而非主动式分配。避免了主动式分配的单点故障问题，当然问题也还是有的，后期我们需要改进的地方主要可以集中在mr_resource_manager结点分配上面，以期达到更有效利用集群资源。