package com.shujia.MapReduce;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;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 Demo05SumScore {//map端public static class MyMapper extends Mapper {@Overrideprotected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {//1500100001,1000001,98String[] splits = value.toString().split(",");String id = splits[0];int score = Integer.parseInt(splits[2]);//以學生id作為key，score作為valuecontext.write(new Text(id),new IntWritable(score));}}//Reduce端public static class MyReducer extends Reducer {@Overrideprotected void reduce(Text key, Iterable values, Reducer.Context context) throws IOException, InterruptedException {//key 學生id//values 每個學生的六門科目成績int sum =0;//記錄總分for (IntWritable score : values) {sum+=score.get();}context.write(key,new IntWritable(sum));}}//Driver端public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://master:9000");//創(chuàng)建一個MapReduce的jobJob job = Job.getInstance(conf);//配置任務job.setJobName("Demo05SumScore");//設置任務運行哪個類job.setJarByClass(Demo05SumScore.class);//配置map端//指定map運行時哪一個類job.setMapperClass(MyMapper.class);//配置Map端輸出的key類型job.setMapOutputKeyClass(Text.class);//配置Map端輸出的value類型job.setMapOutputValueClass(IntWritable.class);//配置Reduce端//指定Reduce運行時哪一個類job.setReducerClass(MyReducer.class);//配置Reduce端輸出的key類型job.setOutputKeyClass(Text.class);//配置Reduce端輸出的value類型job.setOutputValueClass(IntWritable.class);//配置輸入輸出路徑FileInputFormat.addInputPath(job,new Path("/data/score/input"));Path path = new Path("/data/sumScore/output");FileSystem fs = FileSystem.get(conf);//判斷輸出路徑是否存在，存在則刪除if (fs.exists(path)){fs.delete(path,true);}//輸出路徑已存在，會報錯FileOutputFormat.setOutputPath(job,path);//等待任務完成job.waitForCompletion(true);}} package com.shujia.MapReduce;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.IntWritable;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 Demo05SumScore {//map端public static class MyMapper extends Mapper {@Overrideprotected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {//1500100001,1000001,98String[] splits = value.toString().split(",");String id = splits[0];int score = Integer.parseInt(splits[2]);//以學生id作為key，score作為valuecontext.write(new Text(id),new IntWritable(score));}}//Combiner端 發(fā)生在Map端的Reducepublic static class MyCombiner extends Reducer {@Overrideprotected void reduce(Text key, Iterable values, Reducer.Context context) throws IOException, InterruptedException {//key 學生id//values 每個學生的六門科目成績int sum =0;//記錄總分for (IntWritable score : values) {sum+=score.get();}context.write(key,new IntWritable(sum));}}//Reduce端public static class MyReducer extends Reducer {@Overrideprotected void reduce(Text key, Iterable values, Reducer.Context context) throws IOException, InterruptedException {//key 學生id//values 每個學生的六門科目成績int sum =0;//記錄總分for (IntWritable score : values) {sum+=score.get();}context.write(key,new IntWritable(sum));}}//Driver端public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {Configuration conf = new Configuration();conf.set("fs.defaultFS", "hdfs://master:9000");//創(chuàng)建一個MapReduce的jobJob job = Job.getInstance(conf);//配置任務job.setJobName("Demo05SumScore");//設置任務運行哪個類job.setJarByClass(Demo05SumScore.class);//配置map端//指定map運行時哪一個類job.setMapperClass(MyMapper.class);//配置Map端輸出的key類型job.setMapOutputKeyClass(Text.class);//配置Map端輸出的value類型job.setMapOutputValueClass(IntWritable.class);//配置Combinerjob.setCombinerClass(MyCombiner.class);//配置Reduce端//指定Reduce運行時哪一個類job.setReducerClass(MyReducer.class);//配置Reduce端輸出的key類型job.setOutputKeyClass(Text.class);//配置Reduce端輸出的value類型job.setOutputValueClass(IntWritable.class);//配置輸入輸出路徑FileInputFormat.addInputPath(job,new Path("/data/score/input"));Path path = new Path("/data/sumScore/output");FileSystem fs = FileSystem.get(conf);//判斷輸出路徑是否存在，存在則刪除if (fs.exists(path)){fs.delete(path,true);}//輸出路徑已存在，會報錯FileOutputFormat.setOutputPath(job,path);//等待任務完成job.waitForCompletion(true);}}