hadoop作业调度详解之JobClient

4.Jobclient

4.0 配置一个作业

在编写MapReduce程序时通常是上是这样写的:

Job job=new Job(conf,"ClosedDataCube");

//指定作业代码

job.setJarByClass(ClosedDataCube.class);

//设置map

job.setMapperClass(cubeMapper.class);

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(Text.class);

//设置combiner

job.setCombinerClass(cubeCombiner.class);

//设置reduce

job.setReducerClass(cubeReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(Text.class);

//设置输入输出

FileInputFormat.addInputPath(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job,new Path(args[1]));

System.exit(job.waitForCompletion(true)?1:0);

类Job：以作业提交者得角看作业的状况。允许用户配置作业、提交作业、控制执行、查询状态。所有的set方法在提交作业时开始执行。

4.1 提交作业

1.job.waitForCompletion(true)：

提交作业并等待其执行结束。在这里主要通过submit()方法提交一个作业。

2.submit()方法的流程:

RunningJob info=jobClient.submitJobInternal(conf); 通过JobClient提交一个作业。同时返回一个RunningJob 对象用于用户查询这个作业的相信信息。RunningJob是一个接口只有一个实现是JobClient中的NewWorkedJob描述一个job的详细信息。此时 job的状态为运行状态。

3.JobClient.submitJobInternal(conf)

JobClient是用户与JobTracker联系的一个基本的接口。提供提交作业的基本接口、跟踪作业、获得作业的运行状态信息等。

0.创建作业的JobID并提交三个文件

job.xml: 作业配置，例如Mapper, Combiner, Reducer的类型，输入输出格式的类型等。

job.jar: jar包,里面包含了执行此任务需要的各种类，比如 Mapper,Reducer等实现。

job.split: 文件分块的相关信息，比如有数据分多少个块，块的大小(默认64m)等。

1.检查作业的输入、输出

2.计算map数目。即计算输入分片的数目。通过InputFormat的getSplits(job)方法获得作业的split并将split序列化封装为RawSplit。返回split数目，也即代表有多个分片有多少个map。

3.向JobTracker的fs中写入job文件

4.jobSubmitClient.submitJob(jobId)真正的提交一个作业。并返回作业的状态对象句柄。

4.jobSubmitClient.submitJob(jobId)

jobSubmitClient是JobSubmissionProtocol的接口的对象。这个接口有两个实现：LocalJobRunner(conf)当mapred-site.xml中的mapred.job.tracker值为local是为此对象。表示在单机上执行；如果为一个地址的话则是 JobTracker的对象，表示分布式执 行。这里讲JobTracker的处理流程。

jobFile的提交过程是通过RPC(远程进程调用)模块来实现的。大致过程是，JobClient类中通过RPC实现的Proxy接口调用创建了JobTracker对象。与master取得联系。并调用的submitJob()方法提交。JobTracker创建job成功后会给JobClient传回一个JobStatus对象 用于记录job的状态信息，如执行时间、Map和Reduce任务完成的比例等。JobClient会根据这个JobStatus对象创建一个NetworkedJob的RunningJob对象，用于定时从JobTracker获得执行过程的统计数据来监控并打印到用户的控制台。