-
一些复杂的任务很难由一个MR处理完成，所以一般需要将其拆分成为多个简单的MR子任务来执行。
-
MapReduce框架中对于这类的问题提供了几种方式进行任务执行流程的控制，主要包括以下几种方式：
-
顺序组合式MapReduce任务
-
前一个执行完，后面再执行
-
-
依赖关系组合式MapReduce任务
-
前面有多个执行完，后面再执行
-
-
链式MapReduce任务
-
在Map之前或者Reduce之后增加处理
-
-
其中顺序组合式MapReduce任务可以经过变形成为迭代式的MapReduce任务。
-
顺序组合式MapReduce任务
-
多个MR任务作为一个串进行执行，前一个MR的输出作为后一个MR的输入，自动的完成顺序化的执行。
-
顺序组合式MR中：
-
每一个子任务都需要专门的设置独立的配置代码（其实就是和普通的MR是一样的）
-
安装任务的执行顺序设置job的运行顺序
-
任务完成后所有的中间结果输出目录都可以进行删除操作。
-
-
格式: MapReduce1 -> MapReduce2 -> MapReduce3....，
-
每个子任务都必须调用job.waitForCompletion(true)等待job执行完成才可以--就是前一个执行完成，后面才可以开始执行。
优点：结构简单，容易实现。
缺点：由于后一个MR任务必须等待前一个MR任务完成才可以进行，导致集群利用率不高；无法实现多重依赖的关系(或者说多依赖关系的实现比较麻烦)。
package com.rzp.linemr; import org.apache.hadoop.mapreduce.Job; import java.io.IOException; //测试组合mr任务的案例：wordcount案例的输出结果是按照keyword字典排序进行输出,修改成按出现次数排序 public class Demo1 { public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException { //按顺序创建job1,job2... Job job1 = createJobByTimes(1); Job job2 = createJobByTimes(2); //开始执行，这里执行顺序一定要和组合MR的Job的执行顺序一直 runJob(job1,1); runJob(job2,2); System.out.println("Job执行成功"); } //执行Job，执行失败抛出异常 public static void runJob(Job job,int times) throws InterruptedException, IOException, ClassNotFoundException { if (!job.waitForCompletion(true)){ throw new RuntimeException("第"+times+"个job执行失败"); } } /** * 创建job根据给定的参数times * times = 组合式MR任务中第几个job */ public static Job createJobByTimes(int times){ //和普通MR的Job创建一样，从InputFormant开始到OutputFormat给定 //TODO 创建Job return null; } }
依赖关系组合式
-
Hadoop框架为复杂的数据依赖关系提供了一种组合式MapReduce作业的执行流程机制。
-
其实就是MR3依赖于MR1和MR2的结果，但是MR1和MR2不互相依赖，可以同时进行，而如果用顺序式就做不到。
-
Hadoop通过Job和JobControl类为这些作业提供具体的编程方法。
-
Job除了维护配置信息外，还需要维护子任务的依赖关系
-
JobControl类主要用来控制整个作业的执行过程。JobControl是一个Runnable子类，通过线程来调用start方法进行作业的执行流程控制。
-
-
Job完整类名为: org.apache.hadoop.mapred.jobcontrol.Job
-
JobControl完整类名为: org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl
-
优点：实现相对而言简单，提高了集群利用率。
-
缺点：需要自己实现job执行流管理(job失败后执行流失败等操作)
package com.rzp.linemr; import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob; import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl; import org.apache.hadoop.mapreduce.Job; import java.io.IOException; public class Demo2 { public static void main(String[] args) throws IOException { //调用createControlledJob()把mapreduce.Job（普通的job）转换为可控制的ControlledJob对象 ControlledJob job1 = createControlledJob(createJobByTimes(1)); ControlledJob job2 = createControlledJob(createJobByTimes(2)); ControlledJob job3 = createControlledJob(createJobByTimes(3)); //指定依赖关系--job3依赖于job1和job2 //addDependinJob会返回Boolean,可以用于验证 job3.addDependingJob(job1); job3.addDependingJob(job2); //开始创建job的执行流 JobControl jc = new JobControl("测试依赖关系组合式"); //添加job，没有顺序 jc.addJob(job1); jc.addJob(job2); jc.addJob(job3); //总的job个数 int totalSize = jc.getReadyJobsList().size(); //开始执行job流 //因为继承了Runnable接口，可以直接调用run方法 //jc.run(); //更推荐使用Thread来执行 boolean succeeded = false; //job执行流是否成功执行的标志位 try{ new Thread(jc).start(); while (!jc.allFinished()){ //没有执行完，继续进行 try { Thread.sleep(30000); } catch (InterruptedException e) { e.printStackTrace(); } } }finally { //停止执行 jc.stop(); if(jc.allFinished()&&jc.getSuccessfulJobList().size() == totalSize){ //全部执行完，而且执行成功job个数和总的job个数相等，那么任务job执行陈工 succeeded = true; } } System.out.println("job执行"+(succeeded?"成功":"失败")); } //把mapreduce.Job（普通的job）转换为可控制的ControlledJob对象 public static ControlledJob createControlledJob (Job job) throws IOException { ControlledJob cj = new ControlledJob(job.getConfiguration()); cj.setJob(job);//惊醒设置 return cj; } //和普通MR的Job创建一样，从InputFormant开始到OutputFormat给定 public static Job createJobByTimes(int times){ //TODO 创建Job return null; } }
链式MapReduce
-
前两种方式都会有多个Job的启动和关闭，会消耗资源，而Map和Reduce都涉及IO操作，效率不高，因此可以使用链式MR。
-
一个MR任务可能会有一些前处理和后处理，比如说文档倒序索引中可能前处理需要去掉一些“停用词”，后处理需要将一些同义词进行归并。
-
链式MR：链式Mapper(ChainMapper)和链式Reducer(ChainReducer)来完成这种处理。这种作业的执行流程为:map1-->map2-...-->reducer-->map3-->map4-...
-
链式MR要求一个链路中只能有一个reduce操作，可以有多个map操作。
-
-
优点：对于前处理和后处理有要求的MR任务，减轻了操作，提高了效率。
-
缺点：需要指定额外的参数信息（前两种方式，job本身写法和普通MR是一样的，只是在运行主程序上做了操作，但是这种就不同了）。
-
创建完job后，需要使用hadoop提供的专门类设置链路中的map-reduce执行顺序。
-
使用ChainMapper.addMapper添加Map阶段的mapper，按照添加顺序执行，
-
在Reducer阶段必须先使用ChainReducer.setReducer添加reducer处理类，然后才可以使用ChainReducer.addMapper添加mapper处理类，也是按照添加顺序执行job。
-
package com.rzp.linemr; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.Mapper; import org.apache.hadoop.mapred.lib.ChainMapper; import org.apache.hadoop.mapred.lib.ChainReducer; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import java.io.IOException; public class Demo3 { public static void main(String[] args) throws Exception { Configuration conf1 = new Configuration(); Job job1 = Job.getInstance(conf1,"job1"); job1.setJarByClass(Demo3.class); FileInputFormat.addInputPath(job1,new Path("")); /** * 设置mapper * klass 对应mapper类 * K1, V1, K2, V2---klass对应的输入、输出的类型 * mapperConf mapper使用的编写信息 */ //添加第1个mapper ChainMapper.addMapper(JobConf job, Class<? extends Mapper<K1, V1, K2, V2>> klass, Class<? extends K1> inputKeyClass, Class<? extends V1> inputValueClass, Class<? extends K2> outputKeyClass, Class<? extends V2> outputValueClass, boolean byValue, JobConf mapperConf); //添加第2个mapper ChainMapper.addMapper(JobConf job, Class<? extends Mapper<K1, V1, K2, V2>> klass, Class<? extends K1> inputKeyClass, Class<? extends V1> inputValueClass, Class<? extends K2> outputKeyClass, Class<? extends V2> outputValueClass, boolean byValue, JobConf mapperConf); //添加reducer //输入值和上面的一样 ChainReducer.setReducer(...); //添加reducer后续Mapper //格式也和上面的一样 //注意reducer ChainReducer.addMapper(...); //设置总的输入输出路径 job1.setJarByClass(Demo3.class); //TODO 添加map和reduce的输出key和value的类型 } }