前面介绍了什么是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); } } }
特别说明:以上两个实例都是我亲自运行过的,结果也是正确的。
相关推荐
NULL 博文链接:https://zc985552943.iteye.com/blog/2087950
Hadoop 集群配置详解 Hadoop_Hadoop集群(第1期)_CentOS安装配置 Hadoop_Hadoop集群(第2...Hadoop_Hadoop集群(第9期)_MapReduce初级案例 Hadoop_Hadoop集群(第10期)_MySQL关系数据库 Web(Json-Lib类库使用手册)
3.细细品味Hadoop_Hadoop集群(第3期)_VSFTP安装配置 4.细细品味Hadoop_Hadoop集群(第4期)_SecureCRT使用 5.细细品味Hadoop_Hadoop集群(第5期)_Hadoop安装配置 6.细细品味Hadoop_Hadoop集群(第5期副刊)_JDK和...
hadoop2.X配置详解和mapreduce详解
大数据Hadoop核心模块之MapReduce,文档有概念、案例、代码,Mapreduce中,不可多得文档!
使用和学习过老Hadoop框架(0.20.0及之前版本)的同仁应该很熟悉如下的原MapReduce框架图:图1.Hadoop原MapReduce架构从上图中可以清楚的看出原MapReduce程序的流程及设计思路:可以看得出原来的map-reduce架构是...
英特尔Hadoop发行版开发者指南,主要介绍mapreduce、hbase等源代码的详解。
云计算HADOOP、mapreduce、hdfs详解,有兴趣的可以一起研究!
3)Hadoop集群_第3期_VSFTP安装配置_V1.0 4)Hadoop集群_第4期_SecureCRT使用_V1.0 5)Hadoop集群_第5期_Hadoop安装配置_V1.1 6)Hadoop集群_第5期副刊_JDK和SSH无密码配置_V1.0 7)Hadoop集群_第6期_...
Hadoop中mapReduce处理过程详解
024 Hadoop相关命令中的【–config configdir】作用 025 Hadoop 目录结构 026 Eclipse导入Hadoop源码项目 027 HDFS 设计目标 028 HDFS 文件系统架构概述 029 HDFS架构之NameNode和DataNode 030 HDFS 架构讲解总结 ...
如果想将mapreduce结果排序,需将排序对象作为键值。 案例:将利润求和后按照顺序排序 数据源 profit.txt 编号 | 姓名 | 收入 | 支出 1 ls 2850 100 2 ls 3566 200 3 ls 4555 323 1 zs 19000 2000 2 zs 28599 3900 3...
《大数据技术丛书:Hadoop应用开发技术详解》共12章。第1~2章详细地介绍了Hadoop的生态系统、关键技术以及安装和配置;第3章是MapReduce的使用入门,让读者了解整个开发过程;第4~5章详细讲解了分布式文件系统HDFS...
hadoop海量数据处理技术详解,包括hdfs、MapReduce、hive、sqoop等相关技术和伪代码,代码是使用python语言写的。
通过对Hadoop分布式计算平台最核心的分布式文件系统HDFS、MapReduce处理过程,以及数据仓库工具Hive和分布式数据库Hbase的介绍,基本涵盖了Hadoop分布式平台的所有技术核心。通过这一阶段的调研总结,从内部机理的...
MapReduce是Hadoop提供的一套用于进行分布式计算的模型,本身是Doug Cutting根据Google的<MapReduce: Simplified Data Processing on Large Clusters>仿照实现的。 MapReduce由两个阶段组成:Map(映射)阶段和Reduce...
在Hadoop中,用于执行MapReduce任务的机器角色有两个:一个是JobTracker;另一个是TaskTracker,JobTracker是用于调度工作的,TaskTracker是用于执行工作的。一个Hadoop集群中只有一台JobTracker。 在分布式计算...
MapReduce计算模式详解