高阶MapReduce_1_链接多个MapReduce作业

时间:2021-04-14 05:11:44

链接MapReduce作业

1.      顺序链接MapReduce作业

顺序链接MapReduce作业就是将多个MapReduce作业作为生成的一个自动化执行序列,将上一个MapReduce作业的输出作为下一个MapReduce作业的输入。MapReduce作业的链接就类似于Unix 的管道:

mapreduce – 1 | mapreduce – 2 | mapreduce – 3 | ···

                  顺序链接MapReduce作业的执行过程,就是driver为MapReduce作业创建一个带有配置参数的JobConf对象,并将该对象传递给JobClient.runJob()来启动这个作业。而当JobClient.runJob()运行到作业结尾处时会被阻止,MapReduce作业的链接就会在这个时候调用另外一个MapReduce作业的driver。每个作业的driver都必须创建一个新的JobConf对象,并将其输入路径设置为前一个作业的输出路径。

2.      具有复杂依赖的MapReduce链接

如果两个或者多个MapReduce作业之间存在执行的先后顺序关系的话,那么这个时候就不能用顺序链接。Hadoop如何处理这种MapReduce作业之间的关系的呢,Hadoop通过Job和JobControl类来管理这种(非线性)作业之间的依赖。

Job对象是MapReduce作业的表现形式。Job对象的实例化通过传递一个JobConf对象到作业的构造函数中来实现,也就是说通过public Job(JobConf jobConf){}来实现实例化一个MapReduce作业。Job通过Job1.addDependingJob(Job2)来维护作业间的依赖关系这也意味着Job1在Job2完成之前是不会启动的。

JobControl对象负责管理并监视作业Job的执行。添加作业:addJob()方法。而当所有作业和依赖关系添加完成之后,调用JobControl的run()方法,就会生成一个专门提交作业并监视作业执行的线程,这样就实现了JobControl对Job的管理和监视。

3.      预处理和后处理阶段的链接

Hadoop关于预处理和后处理作业的链接提供了两种解决方案。

方案一:为预处理和后处理步骤各自编写一个MapReduce作业,并将其链接起来。在这些步骤中可以使用IdentityReducer。而在此过程中每一个步骤的中间结果都需要占用I/O和存储资源,所以这种方案是非常低效的,一般情况下Hadoop是不建议使用的。

方案二:自己组合mapper和reducer,也就是说自己写mapper去预先调用所有的预处理步骤,再让reducer调用所有的后处理步骤。在Hadoop中呢,是引入了ChainMapper和ChainReducer类来简化预处理和后处理的构成,生成的作业表达式类似于:MAP+ | REDUCE | MAP+   通过按照这个顺序来执行执行多个mapper来预处理数据,并在reduce之后可选的地按序执行多个mapper来做数据的后处理。

例如:有4个mapper作业和一个reduce作业,顺序如下:

Map1 | Map2| Reduce | Map3 | Map4

在这个组合中呢,我们就可以把Map2和Reduce来作业MapReduce作业的核心,把Map1作业预处理步骤,把Map3和Map4作业后处理步骤。

...
conf.setJobName("chain");
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);


JobConf mapAConf = new JobConf(false); ...ChainMapper.addMapper(conf, AMap.class, LongWritable.class, Text.class,Text.class, Text.class, true, mapAConf);

JobConf mapBConf = new JobConf(false); ...ChainMapper.addMapper(conf, BMap.class, Text.class, Text.class,LongWritable.class, Text.class, false, mapBConf);

JobConf reduceConf = new JobConf(false); ...ChainReducer.setReducer(conf, XReduce.class, LongWritable.class, Text.class,Text.class, Text.class, true, reduceConf);

ChainReducer.addMapper(conf, CMap.class, Text.class, Text.class,LongWritable.class, Text.class, false, null);

ChainReducer.addMapper(conf, DMap.class, LongWritable.class,Text.class, LongWritable.class, LongWritable.class, true, null);

FileInputFormat.setInputPaths(conf, inDir);FileOutputFormat.setOutputPath(conf, outDir); ...

JobClient jc = new JobClient(conf); RunningJob job =jc.submitJob(conf); ...


 

 

值得注意的是:在ChainMapper.addMapper()方法的签名中:

publicstatic <k1,v1,k2,v2> void addMapper(JobConf job,

                      Class<? extends Mapper<k1,v1,k2,v2> kclass,

                       Class<? extends k1> inputKeyClass,

                       Class<? extends v2> inputValueClass,

                       Class<? extends k2> outputKeyClass,

                       Class<? extends v2> outputValueClass,

                       boolean byValue,

                       JobConf mapperConf)

            第8个参数为boolean byValue,第一个和最后一个是全局和本地jobconf对象,第二个是mapper类,接下来四个mapper使用的类。

byValue参数:true表示值传递,false表示引用传递。

在标准Mapper中,<k2,v2>是采用值传递被洗牌到不同节点上(传递副本),但是目前我们可以将mapper与另一个链接,就在统一个JVM线程执行,就可以采取引用传递。但是一般来说,map1在调用context.write()写出数据后,这些数据是按约定不会更改的。如果引用传递就会破坏约定。但是使用引用传递会提高效率。如果确定数据不会被破坏,可以设置为false,一般安全起见,设置为true即可。。