Hadoop代码分析（四）

阿娇开发978

好了，大概了解了一下从输入到mapper的数据的处理过程，其实我在刚刚接触hadoop的时候很好奇，当你你在命令行输入hadoop jar ******的时候系统到底如何启动job的，诸如mapper是如何获取输入的K/V对的呢？一般的hadoop程序中的map，reduce方法是被谁而调用的呢？现在就来看看咯：
当一个作业开始被提交，job.waitForCompletion(true)，系统开始等待作业的完成，在waitForCompletion方法中，调用了submit方法以提交作业的conf，在job类中，该方法被描述为如下：
 

public void submit() throws IOException, InterruptedException,
ClassNotFoundException {
ensureState(JobState.DEFINE);
setUseNewAPI();
info = jobClient.submitJobInternal(conf);
state = JobState.RUNNING;
}
 
第5行的jobclient是提交作业，追踪jobtracker，在jobclienr方法中，submitJobInternal（），是一个内部提交作业到系统的方法，


，继续追踪该方法，该方法中有Path submitSplitFile = new Path(submitJobDir, "job.split")；


                                                        maps = writeNewSplits(context, submitSplitFile);


啊哈。终于看到一点split的影子了，不过最主要的是第二条语句，我先说明该语句的作用吧，将该split绑定到一个map上，现在解释原因，追踪writeNewSplits（），看到如下代码：

org.apache.hadoop.mapreduce.InputFormat<?,?> input = ReflectionUtils.newInstance(job.getInputFormatClass(), job.getJobConf());
 

利用反射机制，获取到该job的jobconf，以得到该job的InputFormat（这个是将指定输入转换为splits），然后一直ctrl啊ctrl，到了ReflectionUtils中的setJobConf方法，代码清单：


try {
Class<?> jobConfClass =
conf.getClassByName("org.apache.hadoop.mapred.JobConf");
Class<?> jobConfigurableClass =
conf.getClassByName("org.apache.hadoop.mapred.JobConfigurable");
if (jobConfClass.isAssignableFrom(conf.getClass()) &&
jobConfigurableClass.isAssignableFrom(theObject.getClass())) {
Method configureMethod =
jobConfigurableClass.getMethod("configure", jobConfClass);
configureMethod.invoke(theObject, conf);
}
 

我对反射机制不太懂，反正总结了一下，这段代码就是利用该jobConf信息配置该job的InputFormat，以至于InputFormat

转换的输入是该job的指定输入。。这样，指定的输入就通过分块，给map进行分类处理。。

其实在hadoop中，Context类是很重要的，记录了该jobstracker的上下文信息，明天再说吧，手冷死了。。