MapReduce

news/2024/5/20 10:50:33

1.需求

创建一个文件上传到HDFS，统计每个学生的总成绩，文件内容如下：

使用MapReduce

张三英语 80 河南省

张三数学 50 河南省

张三语文 60 河南省

李四英语 90 河南省

李四语文 90 河南省

李四数学 85 河南省

通过结果：

张三 190

李四 265

2.上传到hdfs

3.IDEA代码

添加依赖

<dependencies><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.1.4</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-mapreduce-client-app</artifactId><version>3.1.4</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-yarn-server-resourcemanager</artifactId><version>3.1.4</version></dependency></dependencies>

package com.yh;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 ScoreTest {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.set("fs.defaultFS","hdfs://hadoop10:8020");Job job = Job.getInstance(conf);job.setJarByClass(ScoreTest.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);TextInputFormat.addInputPath(job,new Path("/score.txt"));TextOutputFormat.setOutputPath(job,new Path("/out4"));job.setMapperClass(ScoreMapper.class);job.setReducerClass(ScoreReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);boolean b = job.waitForCompletion(true);System.out.println(b);}// Mapper类static class ScoreMapper extends Mapper<LongWritable, Text,Text, IntWritable> {private Text student = new Text();private IntWritable score = new IntWritable();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String[] parts = value.toString().split("\\s+"); // 假设字段由空白字符分隔if (parts.length >= 4) {student.set(parts[0]);score.set(Integer.parseInt(parts[2]));context.write(student, score);}}}//Reducer类static class ScoreReducer extends Reducer<Text,IntWritable,Text,IntWritable> {@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int sum = 0;for (IntWritable value : values){sum = sum+value.get();}context.write(key,new IntWritable(sum));}}
}