006_hadoop中MapReduce详解_3

tanggang1740

　　前面介绍了什么是MapReduce，然后通过一个简单的例子来说明MapReduce的流程。但都是针对单个Map函数和Reduce函数。在实际业务中可能会很复杂，可能含有多个MapReduce流程配合使用才能得到想要的结果。本节介绍复杂的MapReduce流程
　　1.线性MapReduce Job流
　　线性含义很简答，就是一个一个MapReduce Job依次执行。AMap的输出交给AReduce，AReduce的输出结果交给BMap，BMap的输出交给BReduce……就这样一直下去。
　　实现方式：将每个Job的启动代码设置成只有上一个Job结束之后才执行，然后将Job的输入设置成上一个Job的输出路劲

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = new Job(conf, "average");
job.setJarByClass(TestMR.class);
job.setMapperClass(MyMap.class);
job.setCombinerClass(MyReduce.class);
job.setReducerClass(MyReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path("/user/text.txt")); //Map的输入
FileOutputFormat.setOutputPath(job, new Path("/user/helloMR/success"));//Reduce的输出
job.waitForCompletion(true);
Job job2 = new Job(conf, "average2");
job2.setJarByClass(WordCount.class);
job2.setMapperClass(TokenizerMapper.class);
job2.setCombinerClass(IntSumReducer.class);
job2.setReducerClass(IntSumReducer.class);
job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job2, new Path("/user/helloMR/success")); //Map的输入
FileOutputFormat.setOutputPath(job2, new Path("/user/helloMR/success2"));//Reduce的输出
System.exit(job2.waitForCompletion(true) ? 0 : 1);
}
　　以上代码仅仅是模拟线性MapReduce Job流，并没用实际的业务含义
　　2.各种依赖MapReduce Job流
　　有时候几个MapReduce Job中可能没有上面说的线性关系。可能是AMapReduce+BMapReduce两个的输出结果做为CMapReduce的输入。并且AMapReduce和BMapReduce之间没有任何关系。
　　实现方式：hadoop为我们提供了这种复杂的Job流API，ControlledJob类和JobControl类。先按照正常情况配置各个Job，配置完后在将所有的Job封装到对应的ControlledJob对象中，然后使用ControlledJob的addDependingJob()设置依赖关系，接着在实例化一个JobControl对象，并使用addJob()方法将多有的Job注入JobControl对象中，最后使用JobControl对象的run方法启动Job流

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
//配置作业1
Job job = new Job(conf, "average");
job.setJarByClass(TestMR.class);
job.setMapperClass(MyMap.class);
job.setCombinerClass(MyReduce.class);
job.setReducerClass(MyReduce.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path("/user/text.txt")); //Map的输入
FileOutputFormat.setOutputPath(job, new Path("/user/helloMR/success"));//Reduce的输出
//配置作业2
Job job2 = new Job(conf, "average2");
job2.setJarByClass(WordCount.class);
job2.setMapperClass(TokenizerMapper.class);
job2.setCombinerClass(IntSumReducer.class);
job2.setReducerClass(IntSumReducer.class);
job2.setOutputKeyClass(Text.class);
job2.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job2, new Path("/user/text.txt")); //Map的输入
FileOutputFormat.setOutputPath(job2, new Path("/user/helloMR/success2"));//Reduce的输出
//配置作业3
Job job3 = new Job(conf, "average2");
job3.setJarByClass(WordCount.class);
job3.setMapperClass(TokenizerMapper.class);
job3.setCombinerClass(IntSumReducer.class);
job3.setReducerClass(IntSumReducer.class);
job3.setOutputKeyClass(Text.class);
job3.setOutputValueClass(IntWritable.class);
//***********作业3的输入是作业1和作业2的输出
FileInputFormat.addInputPath(job3, new Path("/user/helloMR/success")); //Map的输入
FileInputFormat.addInputPath(job3, new Path("/user/helloMR/success2"));
FileOutputFormat.setOutputPath(job3, new Path("/user/helloMR/success3"));//Reduce的输出
/**
* 特别说明：
* 配置依赖关系的作用是确保作业3是在作业1和作业2执行完后才执行，利用作业1和作业2的输出作为作业3的输入。
* 所以在配置作业3时，需要将作业1和作业2的输出路劲作为作业3的输入路径。
*/
//配置依赖关系
ControlledJob cj1 = new ControlledJob(conf);
cj1.setJob(job);
ControlledJob cj2 = new ControlledJob(conf);
cj2.setJob(job2);
ControlledJob cj3 = new ControlledJob(conf);
cj3.setJob(job3);
cj3.addDependingJob(cj1);
cj3.addDependingJob(cj2);
//将所有任务添加到JobControl中
JobControl JC = new JobControl("123");
JC.addJob(cj1);
JC.addJob(cj2);
JC.addJob(cj3);
//启动线程
Thread thread = new Thread(JC);
thread.start();
while(true){
if(JC.allFinished()){  
System.out.println(JC.getSuccessfulJobList());  
JC.stop();  
System.exit(0);
}  
if(JC.getFailedJobList().size() > 0){  
System.out.println(JC.getFailedJobList());  
JC.stop();  
System.exit(0);
}  
}
}
　　特别说明：以上两个实例都是我亲自运行过的，结果也是正确的。