MapReduce自定义Key实现获取学生最高成绩 课程设计

package com.guang.task23; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class Problem1 { public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { // 1.获取配置信息 Configuration conf = new Configuration(); // 2.获取job对象 Job job = Job.getInstance(conf); // 3.关联Driver类的jar job.setJarByClass(Problem1.class); // 4.关联Mapper和Reducer的jar job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReducer.class); // 5.设置Mapper输出的kv类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(MyKey.class); // 6.设置最终输出的kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(MyKey.class); String inputPath = "MapReduce/src/main/java/com/guang/task23/input"; String outputPath = "MapReduce/src/main/java/com/guang/task23/output"; FileSystem fs = FileSystem.get(conf); Path p = new Path(outputPath); if (fs.exists(p)) { fs.delete(p, true); } // 7.设置输入和输出路径 FileInputFormat.setInputPaths(job, new Path(inputPath)); FileOutputFormat.setOutputPath(job, new Path(outputPath)); // 8.提交job System.exit(job.waitForCompletion(true) ? 0 : 1); } public static class WordCountMapper extends Mapper<LongWritable, Text, Text, MyKey> { @Override protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, MyKey>.Context context) throws IOException, InterruptedException { String line = value.toString(); String[] words = line.split(","); context.write(new Text(words[1]), new MyKey(words[0], Integer.parseInt(words[2]))); } } public static class WordCountReducer extends Reducer<Text, MyKey, Text, MyKey> { @Override protected void reduce(Text key, Iterable<MyKey> values, Reducer<Text, MyKey, Text, MyKey>.Context context) throws IOException, InterruptedException { MyKey outV = null; int grade = Integer.MIN_VALUE; for (MyKey value : values) { if (value.getGrade() > grade) { outV = new MyKey(value.getSno(), value.getGrade()); grade = value.getGrade(); } } context.write(key, outV); } } }