MapReduce中Job的初始化

原文：http://langyu.iteye.com/blog/909170

 

可参考：

 http://younglibin.iteye.com/blog/1930169

 

MapReduce依赖Hadoop FileSystem存储job执行过程中需要的所有资源文件。这些文件有job的jar文件、job的配置文件、job的mapper需要处理的目标文件(输入文件)以及job的输出结果。MapReduce可以根据配置文件中File System的URI判断当前是使用哪种Hadoop支持的File System，默认是local system。我更关注job在TT上的表现，而TT又是依赖于DN，所以之后所说的File System都是指HDFS。 

        运行在Cluster上的MapReduce job需要关注的配置文件有：mapred-default.xml与mapred-site.xml，它们之间没有太大的区别，从名称上分，site文件中应当配置与Cluster有关的内容，default就可以随便配置了。与它们有关的引文有：How To Configure 和 Cluster setup。 

               

               
         上图表示job的完整执行流程。本篇blog只关注从第一步到第四步的具体实现，当然也会从模拟的例子按步就班叙述。下面开始我们的进程。。。 

         MapReduce自带WordCount的例子，如流程第一步，在设置基本的参数后，启动job 

              Cluster cluster = new Cluster(config);
Job job = Job.getInstance(cluster);

job.setMapperClass(WordMapper.class);
job.setReducerClass(WordReducer.class);
job.setJarByClass(WordCount.class);
job.setCombinerClass(WordReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

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

job.submit();

                       
        注意这里的FileInputFormat.addInputPath(job, path)，首先得确定MapReduce的输入文件或目录应该在File system上存在，如果MapReduce依赖于HDFS，就得先将本地的文件上传到HDFS上。MapReduce为了防止一个job的输出结果覆盖之前job的输出结果，要求每个job的输出目录都必须独立与其它job，且这个目录在job初始化时不应该存在，只有job需要时才去创建，否则就会报错。我以为MapReduce会在用户第一次设置输出目录时去检查这个目录的有效性，但事实上它是等做了一大堆事情后才去检查，这点让我很困惑。 
      
        在job提交初期，也如流程中的第二步，client会向JT申请一个jobID来作为job的标识符。jobID的格式如job_201101281410_0001，中间的字符串为JT的标识符，后面是job的序号，从1开始一直递增。

        在得到jobID后，MapReduce就需要将job执行必要的资源文件copy到File system上去。在copy之前，我们得先确定这些资源文件存放在File system的什么地方。JT设置有一个工作目录(Staging area, 也称数据中转站)，用来存储与每个job相关的数据。这个目录的前缀由mapreduce.jobtracker.staging.root.dir 参数来指定，默认是/tmp/hadoop/mapred/staging，每个client user可以提交多个job，在这个目录后就得附加user name的信息。所以这个工作目录(Staging area)类似于：/tmp/hadoop/mapred/staging/denny/.staging/。与job相关的资源文件存储的目录是工作目录+jobID：${Staging area}/job_201101281410_0001。 

        这些资源文件存储的情况如下： 
    

stagingArea/job_yyyyMMddHHmm_tttt/job.jar  执行job任务的那个jar文件
stagingArea/job_yyyyMMddHHmm_tttt/files  存储job的输入文件
stagingArea/job_yyyyMMddHHmm_tttt/libjars 与job相关的其它jar文件
stagingArea/job_yyyyMMddHHmm_tttt/archives  job的archives文件

             如果当前的File system是HDFS，那么对于上面的每个文件，我们会设置它在HDFS的replication，这个值由mapreduce.client.submit.file.replication参数指定，默认是10，比普通HDFS文件的默认幅本数大很多，可能也是考虑到把输入数据放到更多的DT上，尽可能实现本地数据计算。 

         把资源文件上传到File system之后，负责job提交的程序会检查job设置的输出目录(output dir)。如果这个目录没有指定或是目录在File system上存在，就会抛出异常。为啥非要在上传那么多文件后才做这项关键检查呢？ 

        接下来才是整个job提交过程中最重要的一步：对输入文件做数据分片(input split)。MapReduce过程中每个mapper怎样知道处理输入文件的哪部分内容呢？理应在mapper执行之前就确定它处理数据的范围吧，那现在的数据分片工作就是干这种事的。更主要的是分片的数量决定map task的数量，它们之间一一对应。这种数据分片(split)只是逻辑分片，记录它应当访问哪个block，及在这个block上的起始index及数据长度的信息。 

        下面我们细说怎样划分数据分片。job可能会有多个输入文件，或许分布在不同的目录下。我们获取输入目录的设置，然后识别得到我们需要处理的那些文件。这里我们可以设置一个PathFilter来过滤那些目录中的文件是否符合我们的要求，自定义的PathFilter类可由mapreduce.input.pathFilter.class属性来设置。对于我们获取的每一个输入文件，根据它的block信息产生数据分片，文件之间不能产生分片。我们可以设置数据分片的数据大小，最小字节数由mapreduce.input.fileinputformat.split.minsize设置，默认是1，最大字节数由mapreduce.input.fileinputformat.split.maxsize设置，默认是Long.MAX_VALUE。由用户定义的分片大小的设置及每个文件block大小的设置，可以计算得分片的大小。计算分片大小的公式是 

splitSize = Math.max(minSize, Math.min(maxSize, blockSize))

从公式可以看出，如果maxSize设置大于blockSize，那么每个block就是一个分片，否则就会将一个block文件分隔为多个分片，如果block中剩下的一小段数据量小于splitSize，还是认为它是独立的分片。 

        产生分片后我们要把这些数据保存起来，序列化到stagingArea/job_yyyyMMddHHmm_tttt/job.split文件中，之后在map task运行时才可以访问到。同时为每个分片分成一个MetaData信息，这个MetaData信息包含每个分片是放在哪台slave server上，它是由JT访问，且作为有效分发map task到拥有物理文件的那台slave server的依据。MetaData信息保存于stagingArea/job_yyyyMMddHHmm_tttt/job.splitmetainfo文件中。 
       
        至此，job提交所需要准备的数据大都已经就绪，前面一步的分片任务也确定了需要多少个map task，与job相关的配置都已确定。把job的配置文件上传到stagingArea/job_yyyyMMddHHmm_tttt/job.xml文件中，在client端做的任务就完成了。Client尝试与JT通信，然后把job提交到JT。