根據提供的數據文件【test.log】

數據文件格式：姓名,語文成績,數學成績,英語成績

完成如下2個案例：

（1）求每個學科的平均成績

（2）將三門課程中任意一門不及格的學生過濾出來

（1）求每個學科的平均成績

• 上傳到hdfs

Idea代碼：

package zz;import demo5.Sort1Job;
import org.apache.hadoop.conf.Configuration;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import java.io.IOException;public class ScoreAverageDriver {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {Configuration conf = new Configuration();conf.set("fs.defaultFS","hdfs://hadoop10:8020");Job job = Job.getInstance(conf);job.setJarByClass(ScoreAverageDriver.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);TextInputFormat.addInputPath(job,new Path("/test.log"));TextOutputFormat.setOutputPath(job,new Path("/test1"));job.setMapperClass(ScoreAverageMapper.class);job.setReducerClass(ScoreAverageReducer.class);//map輸出的鍵與值類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//reducer輸出的鍵與值類型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);boolean b = job.waitForCompletion(true);System.out.println(b);}static class ScoreAverageMapper extends Mapper<LongWritable, Text, Text, IntWritable> {// 定義一個Text類型的變量subject，用于存儲科目名稱private Text subject = new Text();// 定義一個IntWritable類型的變量score，用于存儲分數private IntWritable score = new IntWritable();// 重寫Mapper類的map方法@Overrideprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {// 將輸入的Text值轉換為字符串，并按逗號分割成數組String[] fields = value.toString().split(",");// 假設字段的順序是：姓名,語文成績,數學成績,英語成績String name = fields[0]; // 提取姓名int chinese = Integer.parseInt(fields[1]); // 提取語文成績int math = Integer.parseInt(fields[2]); // 提取數學成績int english = Integer.parseInt(fields[3]); // 提取英語成績// 為Chinese科目輸出成績subject.set("Chinese"); // 設置科目為Chinesescore.set(chinese); // 設置分數為語文成績context.write(subject, score); // 寫入輸出// 為Math科目輸出成績subject.set("Math"); // 設置科目為Mathscore.set(math); // 設置分數為數學成績context.write(subject, score); // 寫入輸出// 為English科目輸出成績subject.set("English"); // 設置科目為Englishscore.set(english); // 設置分數為英語成績context.write(subject, score); // 寫入輸出}}static class ScoreAverageReducer extends Reducer<Text, IntWritable, Text, IntWritable> {// 定義一個IntWritable類型的變量average，用于存儲平均分數private IntWritable average = new IntWritable();// 重寫Reducer類的reduce方法@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {int sum = 0; // 初始化分數總和為0int count = 0; // 初始化科目成績的個數為0// 遍歷該科目下的所有分數for (IntWritable val : values) {sum += val.get(); // 累加分數count++; // 計數加一}// 如果存在分數（即count大于0）if (count > 0) {// 計算平均分并設置到average變量中average.set(sum / count);// 寫入輸出，鍵為科目名稱，值為平均分數context.write(key, average);}}}}

• 結果：

?

（2）將三門課程中任意一門不及格的學生過濾出來

• ?Idea代碼

package zz;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import java.io.IOException;public class FailingStudentDriver {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {Configuration conf = new Configuration();conf.set("fs.defaultFS","hdfs://hadoop10:8020");Job job = Job.getInstance(conf);job.setJarByClass(FailingStudentDriver .class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);TextInputFormat.addInputPath(job,new Path("/test.log"));TextOutputFormat.setOutputPath(job,new Path("/test2"));job.setMapperClass(FailingStudentMapper.class);//map輸出的鍵與值類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setNumReduceTasks(0);boolean b = job.waitForCompletion(true);System.out.println(b);}// 定義一個靜態(tài)類FailingStudentMapper，它繼承了Hadoop的Mapper類
// 該Mapper類處理的是Object類型的鍵和Text類型的值，并輸出Text類型的鍵和NullWritable類型的值static class FailingStudentMapper extends Mapper<Object, Text, Text, NullWritable> {// 定義一個Text類型的變量studentName，用于存儲不及格的學生姓名private Text studentName = new Text();// 定義一個NullWritable類型的變量nullWritable，由于輸出值不需要具體的數據，所以使用NullWritableprivate NullWritable nullWritable = NullWritable.get();// 重寫Mapper類的map方法，這是處理輸入數據的主要方法@Overrideprotected void map(Object key, Text value, Mapper<Object, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {// 將輸入的Text值轉換為字符串，并按逗號分割成數組// 假設輸入的Text值是"姓名,語文成績,數學成績,英語成績"這樣的格式String[] fields = value.toString().split(",");// 從數組中取出學生的姓名String name = fields[0];// 從數組中取出語文成績，并轉換為整數int chineseScore = Integer.parseInt(fields[1]);// 從數組中取出數學成績，并轉換為整數int mathScore = Integer.parseInt(fields[2]);// 從數組中取出英語成績，并轉換為整數int englishScore = Integer.parseInt(fields[3]);// 檢查學生的三門成績中是否有任意一門不及格（即小于60分）// 如果有，則將該學生的姓名寫入輸出if (chineseScore < 60 || mathScore < 60 || englishScore < 60) {studentName.set(name); // 設置studentName變量的值為學生的姓名context.write(studentName, nullWritable); // 使用Mapper的Context對象將學生的姓名寫入輸出}}}}

• 結果：

?