MapReduce 典型场景

Posted 2020-09-11 人生，唯有锻炼与读书不能辜负

一、mapreduce多job串联

   1、需求   

一个稍复杂点的处理逻辑往往需要多个 mapreduce 程序串联处理，多 job 的串联可以借助 mapreduce 框架的 JobControl 实现

   2、实例

以下有两个 MapReduce 任务，分别是 Flow 的 SumMR 和 SortMR，其中有依赖关系： SumMR 的输出是 SortMR 的输入，所以 SortMR 的启动得在 SumMR 完成之后 

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job jobsum = Job.getInstance(conf);
jobsum.setJarByClass(RunManyJobMR.class);
jobsum.setMapperClass(FlowSumMapper.class);
jobsum.setReducerClass(FlowSumReducer.class);
jobsum.setMapOutputKeyClass(Text.class);
jobsum.setMapOutputValueClass(Flow.class);
jobsum.setCombinerClass(FlowSumReducer.class);
jobsum.setOutputKeyClass(Text.class);
jobsum.setOutputValueClass(Text.class);
FileInputFormat.setInputPaths(jobsum, "d:/flow/input");
FileOutputFormat.setOutputPath(jobsum, new Path("d:/flow/output12"));

Job jobsort = Job.getInstance(conf);
jobsort.setJarByClass(RunManyJobMR.class);
jobsort.setMapperClass(FlowSortMapper.class);
jobsort.setReducerClass(FlowSortReducer.class);
jobsort.setMapOutputKeyClass(Flow.class);
jobsort.setMapOutputValueClass(Text.class);
jobsort.setOutputKeyClass(NullWritable.class);
jobsort.setOutputValueClass(Flow.class);
FileInputFormat.setInputPaths(jobsort, "d:/flow/output12");
FileOutputFormat.setOutputPath(jobsort, new Path("d:/flow/sortoutput12"));

ControlledJob sumcj = new ControlledJob(jobsum.getConfiguration());
ControlledJob sortcj = new ControlledJob(jobsort.getConfiguration());
sumcj.setJob(jobsum);
sortcj.setJob(jobsort);
// 设置作业依赖关系
sortcj.addDependingJob(sumcj);
JobControl jc = new JobControl("flow sum and sort");
jc.addJob(sumcj);
jc.addJob(sortcj);
Thread jobThread = new Thread(jc);
jobThread.start();
while(!jc.allFinished()){
Thread.sleep(500);
}
jc.stop();
}

二、topn算法实现——自定义GroupComparator

    1、需求

   在统计学生成绩的小项目中，现在有一个需求：
   求出每个班参考学生成绩最高的学生的信息，班级，姓名和平均分

   2、分析

（1）利用“班级和平均分”作为 key，可以将 map 阶段读取到的所有学生成绩数据按照班级 和成绩排倒序，发送到 reduce
（2）在 reduce 端利用 GroupingComparator 将班级相同的 kv 聚合成组，然后取第一个即是最 大值
   3、实现