一个Hadoop程序的流程

此次我们来探究下Hadoop的job提交过程，下面附上源码中job.class的部分描述
可以看出，一个Hadoop程序流程是创建程序（如mapreduce开发），封装job参数，提交job运行至结束；

/*** The job submitter's view of the Job.* Normally the user * 			1.creates the application, * 			2.describes various facets of the job via {@link Job} * 			3.and then submits the job and monitor its progress.* * <p>Here is an example on how to submit a job:</p>* <p><blockquote><pre>*     // Create a new Job*     Job job = Job.getInstance();*     job.setJarByClass(MyJob.class);*     *     // Specify various job-specific parameters     *     job.setJobName("myjob");*     *     job.setInputPath(new Path("in"));*     job.setOutputPath(new Path("out"));*     *     job.setMapperClass(MyJob.MyMapper.class);*     job.setReducerClass(MyJob.MyReducer.class);**     // Submit the job, then poll for progress until the job is complete*     job.waitForCompletion(true);* </pre></blockquote>*/

Debug看过程

我们利用IEDA工具进行Debug，来探究一下，job提交的过程是怎么样的

Debug--step into

    public boolean waitForCompletion(boolean verbose) throws IOException, InterruptedException, ClassNotFoundException {if (this.state == Job.JobState.DEFINE) { //刚提交时，job正常状态就是DEFINE//这个就是提交的关键过程了 继续step into，进入过程this.submit();	}//下面就是提交之后的一些监控和运行信息的打印if (verbose) {this.monitorAndPrintJob();} else {int completionPollIntervalMillis = getCompletionPollInterval(this.cluster.getConf());while(!this.isComplete()) {try {Thread.sleep((long)completionPollIntervalMillis);} catch (InterruptedException var4) {}}}//执行成功后返回return this.isSuccessful();}

整个过程就是submit方法，我们就详细分析this.submit()，过程如下（源码嵌套过多，不一一展示，可自行debug）；

    public void submit() throws IOException, InterruptedException, ClassNotFoundException {/***ensureState()*	1、状态不是DEFINE*		抛出异常，打印信息：Job in state " + this.state + " instead of " + state*	2、状态是RUNNING 且 cluster对象为空*		抛异常，打印信息：Job in state " + this.state + ", but it isn't attached to any job tracker!*/this.ensureState(Job.JobState.DEFINE);/***setUserNewAPI()*其实主要就是一些Hadoop的项目用的是旧API*需要转化（翻译成）新API*/this.setUseNewAPI();/*** 1、cluster == null* 		创建一个集群对象* 			initiallize:*				锁{	*					客户端协议；*						jobTrackAddr == null {return org.apache.hadoop.mapred.LocalJobRunner}//本地模式	*				  }* 2、其它等等条件下*		就是连接集群的配置，初步看就是决定job运行于什么模式/环境下				**/this.connect();/**提交作业*	真正提交 ：return submitter.submitJobInternal(Job.this, Job.this.cluster);提交job的详细信息*	于此处打断点，一直step into，进入到JobStatus submitJobInternal（Job job,Cluster cluster)*		1、验证job输出路径*			从conf得到路径：*				==null ：Output directory not set*			已经存在：*				Output directory 路径 already exists*		2、获取配置信息*		3、创建jobStagingArea，也就是Hadoop目录下的.staging文件夹*		4、如果ip == null ；根据配置文件设置主机名称，ip*		5、创建jobid*		6、提交配置信息，.xml defualt-xxx.xml等*		7、获得需要的lib下jar包*		8、获得切片信息*		9、job需要的所有内容提交的路径 *				输出流--->job.xml*		10、提交，jobState标记为RUNNING，当前客户端不允许再提交*		11、内部运行，进入下一环节（即开始分析时，打印监控和运行信息那个环节）	*/final JobSubmitter submitter = this.getJobSubmitter(this.cluster.getFileSystem(), this.cluster.getClient());this.status = (JobStatus)this.ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {public JobStatus run() throws IOException, InterruptedException, ClassNotFoundException {return submitter.submitJobInternal(Job.this, Job.this.cluster);}});//状态变为RUNNING，结束后依旧会变为DEFINEthis.state = Job.JobState.RUNNING;LOG.info("The url to track the job: " + this.getTrackingURL());}