初学Hadoop,准备用几篇日志来陈述MapReduce job的生命周期中job提交、task分发和task执行,以及JT scheduling策略,job性能参考等方面的知识。通过代码及参考资料想了解job执行的大致细节,期望在以后job性能调优时有所依据。与细节相关的代码参考于Hadoop-0.21.0版本。
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。
转自:http://langyu.iteye.com/blog/909170