MapReduce的典型编程场景2
发表于：2025-02-05 作者：千家信息网编辑
千家信息网最后更新 2025年02月05日，1.MapReduce 多 Job 串联 介绍：一个稍复杂点的处理逻辑往往需要多个 MapReduce 程序串联处理，多 job 的串联可以借助MapReduce 框架的 JobControl 实
千家信息网最后更新 2025年02月05日MapReduce的典型编程场景2
1.MapReduce 多 Job 串联
介绍：一个稍复杂点的处理逻辑往往需要多个 MapReduce 程序串联处理，多 job 的串联可以借助MapReduce 框架的 JobControl 实现。
需求：
以下有两个 MapReduce 任务，分别是 Flow 的 SumMR 和 SortMR，其中有依赖关系：SumMR的输出是 SortMR 的输入，所以 SortMR 的启动得在 SumMR 完成之后
这两个程序在：https://blog.51cto.com/14048416/2342024
如何实现两个代码的依赖关系呢?
代码实现（这里只给出多 Job 串联的代码）
public class JobDecy { public static void main(String[] args) { Configuration conf = new Configuration(true); conf.set("fs.defaultFS", "hdfs://zzy:9000"); conf.addResource("core-site.xml"); conf.addResource("hdfs-site.xml"); System.setProperty("HADOOP_USER_NAME", "hadoop"); try { //job1 FlowSum Job job1 = Job.getInstance(conf); job1.setJobName("FlowSum"); //设置任务类 job1.setJarByClass(FlowSum.class); //设置Mapper Reducer Combine job1.setMapperClass(FlowSum.MyMapper.class); job1.setReducerClass(FlowSum.MyReducer.class); job1.setCombinerClass(FlowSum.FlowSumCombine.class); //设置map 和reduce 的输入输出类型 job1.setMapOutputKeyClass(Text.class); job1.setMapOutputValueClass(Text.class); job1.setOutputKeyClass(Text.class); job1.setMapOutputValueClass(Text.class); // 指定该 mapreduce 程序数据的输入和输出路径 Path input1 = new Path("/data/input"); Path output1 = new Path("/data/output"); //一定要保证output不存在 if (output1.getFileSystem(conf).exists(output1)) { output1.getFileSystem(conf).delete(output1, true); //递归删除 } FileInputFormat.addInputPath(job1, input1); FileOutputFormat.setOutputPath(job1, output1); //Job2 FlowSumSort Job job2= Job.getInstance(conf); job2.setJarByClass(FlowSumSort.class); job2.setJobName("FlowSumSort"); job2.setMapperClass(Mapper.class); job2.setReducerClass(Reducer.class); job2.setOutputKeyClass(FlowBean.class); job2.setOutputValueClass(NullWritable.class); // 指定该 mapreduce 程序数据的输入和输出路径 Path input2=new Path("//data/output"); Path output2 =new Path("/data/output1"); //一定要保证output不存在 if(output2.getFileSystem(conf).exists(output2)){ output2.getFileSystem(conf).delete(output2,true); //递归删除 } FileInputFormat.addInputPath(job2,input2); FileOutputFormat.setOutputPath(job2,output2); //为每个任务创建ControlledJob ControlledJob job1_cj=new ControlledJob(job1.getConfiguration()); ControlledJob job2_cj=new ControlledJob(job2.getConfiguration()); //绑定 job1_cj.setJob(job1); job2_cj.setJob(job2); // 设置作业依赖关系 job2_cj.addDependingJob(job2_cj); //job2 依赖于job1 //创建jobControl JobControl jc=new JobControl("sum and sort"); jc.addJob(job1_cj); jc.addJob(job2_cj); //使用线程开启Job Thread jobThread=new Thread(jc); //开启任务 jobThread.start(); //为了保证主程序不终止,没0.5秒检查一次是否完成作业 while(!jc.allFinished()){ try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } } //罪作业完成之后，终止线程，释放资源 jc.stop(); } catch (IOException e) { e.printStackTrace(); } }}2.TopN 算法实现（二次排序）
需求：求每个班级的总分最高的前三名
字段：班级 姓名 数学 语文 英语 （字段之间是制表符分割）
分析：
- 利用"班级和总分"作为 key，可以将 map 阶段读取到的所有学生成绩数据按照班级和成绩排倒序，发送到 reduce
- 在 reduce 端利用 GroupingComparator 将班级相同的 kv 聚合成组，然后取前三个即是前三名
代码实现：
自定义学生类：
public class Student implements WritableComparable { private String t_class; private String t_name; private int t_sumSource; public Student(){ } public void set(String t_class,String t_name,int chinese,int math,int english){ this.t_class=t_class; this.t_name=t_name; this.t_sumSource=chinese+math+english; } public String getT_class() { return t_class; } public void setT_class(String t_class) { this.t_class = t_class; } public String getT_name() { return t_name; } public void setT_name(String t_name) { this.t_name = t_name; } public int getT_sumSource() { return t_sumSource; } public void setT_sumSource(int t_sumSource) { this.t_sumSource = t_sumSource; } //比较规则 @Override public int compareTo(Student stu) { //首先根据班级比较 int result1=this.t_class.compareTo(stu.t_class); //班级相同的在根据总分比较 if(result1==0){ return stu.t_sumSource-this.t_sumSource; } return result1; } //序列化 @Override public void write(DataOutput out) throws IOException { out.writeUTF(this.t_class); out.writeUTF(this.t_name); out.writeInt(this.t_sumSource); } //反序列化 @Override public void readFields(DataInput in) throws IOException { this.t_class=in.readUTF(); this.t_name=in.readUTF(); this.t_sumSource=in.readInt(); }} 自定义分组：
//自定义分组规则 private static class MyGroupComparator extends WritableComparator{ //这句代码必须要加，并且要调用父类的构造 public MyGroupComparator(){ super(Student.class, true); } / 决定输入到 reduce 的数据的分组规则 根据班级进行分组 / @Override public int compare(WritableComparable a, WritableComparable b) { Student stu1=(Student)a; Student stu2=(Student)a; return stu1.getTclass().compareTo(stu2.getTclass()); } }*MR程序：
//Mapper
private static class MyMapper extends Mapper { Student bean = new Student(); NullWritable mv = NullWritable.get(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\\s+"); //班级 姓名 数学 语文 英语 String t_clas=fields[0]; String t_name=fields[1]; int chinese=Integer.parseInt(fields[2]); int math=Integer.parseInt(fields[3]); int english=Integer.parseInt(fields[4]); bean.set(t_clas,t_name,chinese,math,english); context.write(bean,mv); } } //Reducer private static class MyReducer extends Reducer { @Override protected void reduce(Student key, Iterable values, Context context) throws IOException, InterruptedException { int count =0; for(NullWritable value:values){ if(count>2){ break; } context.write(key,value); count++; } } } job：
public class ClazzScoreGroupComparator { public static void main(String[] args) { Configuration conf=new Configuration(true); conf.set("fs.defaultFS","hdfs://zzy:9000"); conf.set("fs.defaultFS", "hdfs://zzy:9000"); conf.addResource("core-site.xml"); conf.addResource("hdfs-site.xml"); System.setProperty("HADOOP_USER_NAME", "hadoop"); try { Job job= Job.getInstance(conf); job.setJarByClass(ClazzScoreGroupComparator.class); job.setJobName("ClazzScoreGroupComparator"); job.setMapperClass(MyMapper.class); job.setReducerClass(MyReducer.class); //指定自定义分组 job.setGroupingComparatorClass(MyGroupComparator.class); job.setOutputKeyClass(Student.class); job.setOutputValueClass(NullWritable.class); // 指定该 mapreduce 程序数据的输入和输出路径 Path input=new Path("//data/student.txt"); Path output =new Path("/data/output2"); //一定要保证output不存在 if(output.getFileSystem(conf).exists(output)){ output.getFileSystem(conf).delete(output,true); //递归删除 } FileInputFormat.addInputPath(job,input); FileOutputFormat.setOutputPath(job,output); boolean success=job.waitForCompletion(true); System.exit(success?0:1); } catch (Exception e) { e.printStackTrace(); } }3. MapReduce 全局计数器
介绍：计数器是用来记录 job 的执行进度和状态的。它的作用可以理解为日志。我们可以在程序的某个位置插入计数器，记录数据或者进度的变化情况，MapReduce 自带了许多默认 Counter，现在我们来分析这些默认 Counter 的含义，方便大家观察 Job 结果，如输入的字节数、输出的字节数、Map 端输入/输出的字节数和条数、Reduce 端的输入/输出的字节数和条数等。
需求：利用全局计数器来统计一个目录下所有文件出现的单词总数和总行数
代码实现：
public class CounterWordCount { public static void main(String[] args) { Configuration conf=new Configuration(true); conf.set("fs.defaultFS","hdfs://zzy:9000"); conf.set("fs.defaultFS", "hdfs://zzy:9000"); conf.addResource("core-site.xml"); conf.addResource("hdfs-site.xml"); System.setProperty("HADOOP_USER_NAME", "hadoop"); try { Job job= Job.getInstance(conf); job.setJarByClass(CounterWordCount.class); job.setJobName("CounterWordCount"); job.setMapperClass(MyMapper.class); //设置reduceTask为0 job.setNumReduceTasks(0); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); // 指定该 mapreduce 程序数据的输入和输出路径 Path input=new Path("//data/"); Path output =new Path("/data/output3"); //一定要保证output不存在 if(output.getFileSystem(conf).exists(output)){ output.getFileSystem(conf).delete(output,true); //递归删除 } FileInputFormat.addInputPath(job,input); FileOutputFormat.setOutputPath(job,output); boolean success=job.waitForCompletion(true); System.exit(success?0:1); } catch (Exception e) { e.printStackTrace(); } } //定义枚举 用于存放计数器 enum CouterWordsCounts{COUNT_WORDS, COUNT_LINES} //Mapper private static class MyMapper extends Mapper { Text mk=new Text(); LongWritable mv=new LongWritable(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 统计行数，因为默认读取文本是逐行读取，所以 map 执行一次，行数+1 context.getCounter(CouterWordsCounts.COUNT_LINES).increment(1); String words[]=value.toString().split("\\s+"); for(String word:words){ context.getCounter(CouterWordsCounts.COUNT_WORDS).increment(1); } } //这个方法，在这个类的最后执行 @Override protected void cleanup(Context context) throws IOException, InterruptedException { mk.set("行数："); mv.set(context.getCounter(CouterWordsCounts.COUNT_LINES).getValue()); context.write(mk,mv); mk.set("单词数："); mv.set(context.getCounter(CouterWordsCounts.COUNT_WORDS).getValue()); context.write(mk,mv); } }} 4.MapReduce Join
介绍：在各种实际业务场景中，按照某个关键字对两份数据进行连接是非常常见的。如果两份数据都比较小，那么可以直接在内存中完成连接。如果是大数据量的呢？显然，在内存中进行连接会发生 OOM。MapReduce 可以用来解决大数据量的连接。在MapReduce join分两种，map join和reduce join
map join
介绍：MapJoin 适用于有一份数据较小的连接情况。做法是直接把该小份数据直接全部加载到内存当中，按链接关键字建立索引。然后大份数据就作为 MapTask 的输入，对 map()方法的每次输入都去内存当中直接去匹配连接。然后把连接结果按 key 输出.。
数据介绍：
movies.dat：1::Toy Story (1995)::Animation|Children's|Comedy
字段含义：movieid, moviename, movietype
Ratings.dat：1::1193::5::978300760
字段含义：userid, movieid, rate, timestamp
代码实现：
public class MovieRatingMapJoinMR { public static void main(String[] args) { Configuration conf = new Configuration(); conf.set("fs.defaultFS", "hdfs://hadoop02:9000"); System.setProperty("HADOOP_USER_NAME", "hadoop"); try { Job job = Job.getInstance(conf); job.setJarByClass(MovieRatingMapJoinMR.class); job.setMapperClass(MyMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setNumReduceTasks(0); String minInput = args[0]; String maxInput = args[1]; String output = args[2]; FileInputFormat.setInputPaths(job, new Path(maxInput)); Path outputPath = new Path(output); FileSystem fs = FileSystem.get(conf); if (fs.exists(outputPath)) { fs.delete(outputPath, true); } FileOutputFormat.setOutputPath(job, outputPath); //将小表加载到内存 URI uri=new Path(minInput).toUri(); job.addCacheFile(uri); boolean status = job.waitForCompletion(true); System.exit(status?0:1); } catch (Exception e) { e.printStackTrace(); } } //Mapper private static class MyMapper extends Mapper { Text mk = new Text(); Text mv = new Text(); // 用来存储小份数据的所有解析出来的 key-value private static Map movieMap = new HashMap(); @Override protected void setup(Context context) throws IOException, InterruptedException { //读取加载到内存的表数据，并将数据的封装到movieMap容器中 URI[] cacheFiles = context.getCacheFiles(); //获取文件名 String myfilePath = cacheFiles[0].toString(); BufferedReader br = new BufferedReader(new FileReader(myfilePath)); // 此处的 line 就是从文件当中逐行读到的 movie String line = ""; while ((line = br.readLine()) != null) { //movieid::moviename::movietype String fields[] = line.split("::"); movieMap.put(fields[0], fields[1] + "\\t" + fields[2]); } IOUtils.closeStream(br); } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("::"); //userid::movieid::rate::timestamp String userid = fields[0]; String movieid = fields[1]; String rate = fields[2]; String timestamp = fields[3]; if (movieMap.containsKey(userid)) { String movieFileds = movieMap.get(userid); mk.set(userid); mv.set(movieFileds + "\\t" + movieid + "\\t" + rate + "\\t" + timestamp); context.write(mk, mv); } } }} reduce join
介绍：
- map 阶段，两份数据 data1 和 data2 会被 map 分别读入，解析成以链接字段为 key 以查询字段为 value 的 key-value 对，并标明数据来源是 data1 还是 data2。
- reduce 阶段，reducetask 会接收来自 data1 和 data2 的相同 key 的数据，在 reduce 端进行乘积链接，最直接的影响是很消耗内存，导致 OOM
数据介绍：
movies.dat：1::Toy Story (1995)::Animation|Children's|Comedy
字段含义：movieid, moviename, movietype
Ratings.dat：1::1193::5::978300760
字段含义：userid, movieid, rate, timestamp
代码实现：
public class MovieRatingReduceJoinMR { public static void main(String[] args) { Configuration conf = new Configuration(); conf.set("fs.defaultFS", "hdfs://zzy:9000"); System.setProperty("HADOOP_USER_NAME", "hadoop"); try { Job job = Job.getInstance(conf); job.setJarByClass(MovieRatingReduceJoinMR.class); job.setMapperClass(MyMapper.class); job.setReducerClass(MyReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); String Input = args[0]; String output = args[1]; FileInputFormat.setInputPaths(job, new Path(Input)); Path outputPath = new Path(output); FileSystem fs = FileSystem.get(conf); if (fs.exists(outputPath)) { fs.delete(outputPath, true); } FileOutputFormat.setOutputPath(job, outputPath); boolean status = job.waitForCompletion(true); System.exit(status?0:1); } catch (Exception e) { e.printStackTrace(); } } //Mapper private static class MyMapper extends Mapper { private String name; Text mk = new Text(); Text mv = new Text(); //获取文件名 @Override protected void setup(Context context) throws IOException, InterruptedException { //InputSplit是一个抽象类，使用它的实现类FileSplit FileSplit is=(FileSplit)context.getInputSplit(); name=is.getPath().getName(); } @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //movies.dat movieid::moviename::movietype //ratings.dat userid::movieid::rate::timestamp String OutputKey=null; String OutputValue=null; String fields[]=value.toString().split("::"); if(name.endsWith("movies.dat")){ OutputKey=fields[0]; OutputValue=fields[1]+"\t"+fields[2]+"_"+"movies"; }else if(name.endsWith("ratings.dat")){ OutputKey=fields[1]; OutputValue=fields[0]+"\t"+fields[2]+"\t"+fields[3]+"_"+"ratings"; } mk.set(OutputKey); mv.set(OutputValue); context.write(mk,mv); } } //Reducer private static class MyReducer extends Reducer< Text, Text, Text, Text>{ Text rv=new Text(); @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { List movies=new ArrayList<>(); List ratings=new ArrayList<>(); //将数据分别添加到存放两张表字段的容器中 for(Text value:values){ String fields[]= value.toString().split("_"); if(fields[1].equals("movies")){ movies.add(fields[0]); }else if(fields[1].equals("ratings")){ ratings.add(fields[0]); } } //连接两个表的数据 if(ratings.size()>0&&movies.size()>0){ for(String movie:movies){ for(String rate:ratings){ rv.set(movie+"\t"+rate); context.write(key,rv); } } } } }}
数据
输入
输出
字段
班级
代码
程序
内存
含义
计数器
保证
分组
两个
任务
字节
文件
路径
递归
相同
总分
数据库的安全要保护哪些东西
数据库安全各自的含义是什么
生产安全数据库录入
数据库的安全性及管理
数据库安全策略包含哪些
海淀数据库安全审计系统
建立农村房屋安全信息数据库
易用的数据库客户端支持安全管理
连接数据库失败ssl安全错误
数据库的锁怎样保障安全
米多科技与武汉互联网
csv数据库如何导入
公安软件开发技术组
广州安卓应用软件开发公司哪家强
十四五规划计算机网络技术专业
蓓优蜜互联网科技公司
tt数据库查看所有表
数据库图片存储的两种主要方式
选择网络安全模式启动后蓝屏
江财数据库试卷
深圳工业软件开发价格
工人数据库管理
网络安全账户借给他人后果
互联网行业微信科技大号
信息网络安全服务平台
服务器代理端安全管理软件
数据库sql修改记录的命令
大城县天气预报软件开发
汽车远程启动需要专门的服务器
球球大作战中心服务器怎么安装
oa系统用什么软件开发
禁止excel修改数据库
服务器主板插槽式供电
计算机网络安全是指
思科网络技术学院靠谱吗
工人数据库管理
服务器如何拆开
万户网络技术有限公司钱贵昱
手机数据库更新软件
天津精英网络技术咨询联系人
