Hadoop代码分析（四）

 

好了，大概了解了一下从输入到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的上下文信息，明天再说吧，手冷死了。。