[hadoop源码阅读][9]-mapreduce-job提交过程

xlfm22

　　1.从wordcount作为入口



public class WordCount
{
public static void main(String[] args) throws Exception
{
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
Job job = new Job(conf, "word count");//job类的主要工作就是设置各种参数
job.setJarByClass(WordCount.class);//将包含WordCount.class的jar找到,后面要上传的
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);//如果传入参数为ture则及时打印作业运作信息，否则只是等待作业结束
}
}


　　
　　2.job.waitForCompletion(true)函数内主要干活的是submit




public void submit() throws IOException, InterruptedException, ClassNotFoundException
{
ensureState(JobState.DEFINE);

//Job.setMapperClass(xxx.class)：实际上设置的是mapreduce.map.class，即New。JobConf.setMapperClass(xxx.class)：实际上设置的是mapred.mapper.class，即Old。可见不调用JobConf.setMapperClass，就应该是使用的NewAPI。


    setUseNewAPI();
info = jobClient.submitJobInternal(conf);//实际的job提交过程.info用来和jobtracker进行交互,对提交的job进行监控以及杀死等操作

state = JobState.RUNNING;
}
　　
　　3.在看jobClient.submitJobInternal(conf)函数之前,先看jobclient这个对象的构造过程:
　　3.1先将mapred-site.xml和core-site.xml包含到conf中.static代码
　　3.2 init函数




public void init(JobConf conf) throws IOException
{
String tracker = conf.get("mapred.job.tracker", "local");//如果没有设置,或者没有找到上面2个xml配置文件
if ("local".equals(tracker))
{
conf.setNumMapTasks(1);//local模式 reduce只能是1
this.jobSubmitClient = new LocalJobRunner(conf);
}
else
{//创建rpc
this.jobSubmitClient = createRPCProxy(JobTracker.getAddress(conf), conf);
}
}
//jobSubmitClient对应的类是JobSubmissionProtocol的实现之一（目前有两个实现，JobTracker和LocalJobRunner）
　　3.3 submitJobInternal函数




public RunningJob submitJobInternal(JobConf job)
{
JobID jobId = jobSubmitClient.getNewJobId();
Path submitJobDir = new Path(getSystemDir(), jobId.toString());//conf.get("mapred.system.dir", "/tmp/hadoop/mapred/system")
Path submitJarFile = new Path(submitJobDir, "job.jar");
Path submitSplitFile = new Path(submitJobDir, "job.split");
/*1.建立submitJobDir目录,2.将参数中指定的jars,files,archives放到分布式缓存中,
* 3.将main函数所在的jar包上传为submitJarFile
* 4.设置user,group,这个对hdfs的文件操作是有权限影响的,设置当前工作目录
* */
configureCommandLineOptions(job, submitJobDir, submitJarFile);
Path submitJobFile = new Path(submitJobDir, "job.xml");
int reduces = job.getNumReduceTasks();
JobContext context = new JobContext(job, jobId);
// 检测输出目录是否存在,如果存在是会报错的
org.apache.hadoop.mapreduce.OutputFormat >List getSplits(JobContext job)
　　3.1---->>List listStatus(JobContext job)//过滤掉输入路径下,以_和.开头的路径,以及根据用户设置的mapred.input.pathFilter.class对文件进行过滤,得到文件列表
　　3.2 如果文件是压缩的,也即是不可splitable的,那么整个文件作为一个split
　　3.3 如果文件是splitable的,那么首先计算每个split的大小,mapred.min.split.size的,默认大小是1,
　　mapred.max.split.size的默认值是Long.MAX_VALUE,blockSize的默认大小是64M,
　　那么split的大小为Math.max(minSize, Math.min(maxSize, blockSize));从公式可以看出，
　　如果maxSize设置大于blockSize，那么每个block就是一个分片，否则就会将一个block文件分隔为多个分片，
　　如果block中剩下的一小段数据量小于splitSize，还是认为它是独立的分片。
　　3.4 将每个split的路径,大小,下标,以及位置信息保存到split数组
　　2.2 安装每个split的大小进行排序,将大的放在前面,然后序列化到文件中.
　　
　　
　　参考文献
　　http://langyu.iteye.com/blog/909170
　　http://blog.iyunv.com/andyelvis/article/details/7706205
　　http://www.iyunv.com/forfuture1978/archive/2010/11/19/1882268.html
　　http://www.iyunv.com/spork/archive/2010/04/21/1717552.html
　　http://baoshengdeer.sinaapp.com/?p=116