Hadoop实战：WordCount与成绩统计详解

2301_80691430

712人浏览 · 2026-03-04 00:30:56

2301_80691430 · 2026-03-04 00:30:56 发布

一、MapReduce概述

MapReduce是⼀个分布式、并⾏处理的计算框架。 MapReduce把任务分为Map阶段和Reduce阶段。开发⼈员使⽤存储在HDFS中数据（可实现快速存储），编写Hadoop的MapReduce任务。由于MapReduce⼯作原理的特性，Hadoop能以并⾏的⽅式访问数据，从⽽实现快速访问数据。

⼆、MapReduce编程基础

1、Hadoop数据类型 Hadoop数据包括：BooleanWritable、ByteWritable、DoubleWritable、FloatWritale、 IntWritable、LongWritable、Text、NullWritable等，它们实现了WritableComparable接⼝。其中 Text表⽰使⽤UTF8格式存储的⽂本、NullWritable类型是当（key,value）中的key或value为空时使用。

除了上述Hadoop类型外，⽤⼾还可以⾃定义新的数据类型。⽤⼾⾃定义数据类型需要实现 Writable接⼝，但如果需要作为主键key使⽤或需要⽐较⼤⼩时，则需要实现WritableComparable接⼝。

2、数据输⼊格式InputFormat 抽象类InputFormat有三个直接⼦类： FileInputFormat、DBInputFormat、DelegatingInputFormat

其中，⽂件输⼊格式类FileInputFormat类有⼏个⼦类： TextInputFormat、KeyValueInputFormat、SequenceFileInputFormat、 NlineInputFormat、CombineFileInputFormat

序列化⽂件输⼊类SequenceFileInputFormat有⼏个⼦类： SequenceFileAsBinaryInputFormat、SequenceFileAsTextInputFormat、 SequenceFileInputFilter

数据库输⼊格式类DBInputFormat的直接⼦类是：DataDriverDBInputFormat，⽽这个⼦类⼜派⽣⼦类：OracleDataDriverDBInputFormat

常⽤数据输⼊格式类
InputFormat类	描述	键（key）	值（value）
TextInputFormat	默认输入格式，读取文本文件的行	当前行的偏移量	当前行内容
KeyValueTextInputFormat	将行解析成键值对	行内首个制表符的内容	行内其余内容
SequenceFileInputFormat	专用于高性能的二进制格式	用户定义	用户定义

3、输⼊数据分块InputSplit和数据记录读⼊RecordReader 编程时由⽤⼾选择的数据输⼊格式InputFormat类型来⾃动决定数据分块InputSplit和数据记录 RecordReader类型。⼀个InputSplit将单独作为⼀个Mapper的输⼊，即作业的Mapper数量是由 InputSplit个数决定的。

4、数据输出格式OutputFormat：

抽象类OutputFormat<K,V>有四个直接⼦类： FileOutputFormat<K,V>、DBOutputFormat<K,V>、NullOutputFormat<K,V>、 FilterOutputFormat<K,V>

FileOutputFormat<K,V>有两个直接⼦类： TextOutputFormat<K,V>、SequenceFileOutputFormat<K,V>

SequenceFileOutputFormat<K,V>有直接⼦类：SequenceFileAsBinaryOutputFormat

FilterOutputFormat<K,V>有直接⼦类：LazyOutputFormat

5、数据记录输出类RecordWriter 数据记录输出类RecordWriter是⼀个抽象类。

6、Mapper类 Mapper类是⼀个抽象类，位于hadoop-mapreduce-client-core-2.x.x.jar中，其完整类名是： org.apache.hadoop.mapreduce.Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>，需派⽣⼦类使⽤，在⼦类中重写map⽅法：map(KEYINkey,VALUEINvalue,Mapper.Contextcontext)对出⼊的数据分块每个键值对调⽤⼀次。

7、Reduce类 Reduce类是⼀个抽象类，位于hadoop-mapreduce-client-core-2.x.x.jar中，其完整类名是： org.apache.hadoop.mapreduce.Reduce<KEYIN,VALUEIN,KEYOUT,VALUEOUT>，需派⽣⼦类使⽤，在⼦类中重写reduce⽅法：reduce(KEYINkey,Inerablevalue,Reducer.Contextcontext) 对出⼊的数据分块每个键值对调⽤⼀次。

三、开发环境配置

⾸先在Eclipse中创建⼀个Java⼯程，项⽬的编译环境要求在jdk1.8以上即可；创建一个叫lib的文件夹放入以下架包：hadoop/share/hadoop/common/⽬录下的所有jar包，以及其⼦⽬录lib下的jar包；hadoop/share/hadoop/hdfds/⽬录下的所有jar包和其⼦⽬录下的所有jar包；hadoop/share/hadoop/mapreduce/⽬录下的所有jar包；在创建好的⼯程上点击右键，选择BuildPath-->ConfigureBuildPath..

四、WordCount程序实现流程

分别编写Mapper、Reducer和Job任务三段程序：

Mapper：

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.io.LongWritable;
public class WordCountMapper extends Mapper<Object, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context) 
            throws IOException, InterruptedException {
        StringTokenizer itr = new StringTokenizer(value.toString());
        while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            context.write(word, one);
        }
    }
}

Reducer：

import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
	private IntWritable result = new IntWritable();
	public void reduce(Text key, Iterable<IntWritable> values, Context context) 
        throws IOException, InterruptedException {
			int sum = 0;
			IntWritable val;
			for(Iterator<IntWritable> i$ = values.iterator(); i$.hasNext(); sum +=val.get()) {
				val = i$.next();
		}
		this.result.set(sum);
		context.write(key, this.result);
	}
}

Job任务：

import java.io.IOException;
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;
public class WordCountMapReduce {
	public static void main (String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		String inPath="hdfs://192.168.74.138:9000/input.txt";
		String outPath="hdfs://192.168.74.138:9000/output/";
		Configuration conf = new Configuration();
		Job job = Job.getInstance();
		job.setJarByClass(WordCountMapReduce.class);
		job.setMapperClass(WordCountMapper.class);
		job.setCombinerClass(WordCountReducer.class);
		job.setReducerClass(WordCountReducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job,new Path(inPath)); 
		FileOutputFormat.setOutputPath(job,new Path(outPath));
		boolean b = job.waitForCompletion(true);
		if(!b) {
			System.out.println("wordcount 任务失败");
		}
	}
}

编译运行后打包输出：因为MapReduce程序是在Hadoop平台上运⾏的，所以在本地⽆法运⾏，需要编译打包后上传到Hadoop服务器运⾏。

在项⽬右键单击，选择RunAs-->JavaApplication通过在搜索栏中输⼊主类的名称来查找对应的主类，这⾥注意主类名称及项⽬名称，不要搞错。点击OK后进⾏编译并运⾏，运行时虽然出了错，但是没有关系。提⽰是说在运⾏时没有提供需要的主类参数。

以上步骤完成后，开始打包输出。在项⽬上单击右键，选择Export，选择Java⽬录下的RunnableJARfile这⼀项。在对话框中依次选择项⽬的主类以及输出的⽬录及输出的⽂件名称，类型为jar包⽂件。

上⾯注意必须选择ExtractrequiredlibrariesintogeneratedJAR这⼀项。单击Finish，在后续弹出的对话框中均选择OK即可完成打包输出。

上传⽂件到Hadoop服务器中并运⾏。使用winscp工具将将⽂件复制到Hadoop的服务的⽤⼾主⽬录下。登录Hadoop服务器上并启动Hadoop集群。

如上图所⽰，则hadoop集群启动完毕。检查当前hadoop服务器下的⽬录及⽂件情况。如果没有input⽬录则要创建⼀个input⽬录到当前⽤⼾⽬录下。并且将output⽬录删除。

创建input.txt文件，在这⽂件中分别写⼊以下的内容：

将input.txt⽂件上传到hadoop服务器中的input⽬录下，以上步骤为程序前期准备，下⾯运⾏程序并查看结果。执⾏下⾯的命令，运⾏程序

使⽤下⾯的命令查看运⾏结果：hadoop@hadoop:~$ hadoop jar wordcount.jar WordCountMapReduce /input/input.txt /output

使⽤下⾯的命令查看运⾏结果：hadoop@hadoop:~$ hdfs dfs -cat /output/*

五、考试平均成绩计算流程

准备多科目数据，创建三个成绩文件：

将⽂件上传到HDFS的data⽬录下：

编写Mapper类

package hadoop;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Mapper;
public class CourseScoreAverageMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
	throws IOException, InterruptedException {
		String line = new String(value.getBytes(), 0, value.getLength(), "UTF8");
		Counter countPrint = context.getCounter("CourseScoreAverageMapper.Map 输出传递Value", line);
		countPrint.increment(1L);
		StringTokenizer tokenArticle = new StringTokenizer(line, "\n");
		while(tokenArticle.hasMoreElements()) {
			StringTokenizer tokenLine = new StringTokenizer(tokenArticle.nextToken());
			String strName = tokenLine.nextToken();
			String strScore = tokenLine.nextToken();
			Text name = new Text(strName);
			int scoreInt = Integer.parseInt(strScore);
			context.write(name, new IntWritable(scoreInt));
			countPrint = context.getCounter("CourseScoreAverageMapper.Map中循环输出信息：", "<key.value>:<" + strName + "，" + strScore + ">");
			countPrint.increment(1L);
		}
	}		
}

编写Reducer类（含Combiner）

package hadoop;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Reducer;
import org.mockito.internal.stubbing.answers.ThrowsException;
public class CourseScoreAverageReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
	protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) 
	throws IOException, InterruptedException {
		int sum = 0;
		int count = 0;
		for(IntWritable val:values) {
			sum += val.get();
			count++;
		}
		int average = (int)sum/count;
		context.write(key, new IntWritable(average));
		Counter countPrint = context.getCounter("CourseScoreAverageReducer.Reducer中输出信息：", "<key,value>:<" + key.toString() + "," + average + ">");
		countPrint.increment(1L);
	}
}

配置Job并启用Combiner

package hadoop;
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.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class CourseScoreDriver{
	public static void main(String[] args) throws Exception{
		Configuration conf=new Configuration(); //获取配置⽂件
		Job job=Job.getInstance(conf,"CourseScoreAverage"); 
		String[] otherArgs=new GenericOptionsParser(conf,args).getRemainingArgs(); //获取命令⾏参数
		if(otherArgs.length<2){ 
			System.err.print("Usage:hadoopjarMyAverage.jar<in><out>");
			System.err.print("hadoopjarMyAverage.jarhadoop.mapreduce.CourseScoreDriver<in><out>");
			System.exit(2);
		}
		else{
			for(int i=0;i<otherArgs.length-1;i++){ //设置⽂件输⼊路径
	if(!("hadoop.mapreduce.CourseScoreDriver".equalsIgnoreCase(otherArgs[i]))){
					FileInputFormat.addInputPath(job,new Path(otherArgs[i]));
					System.out.println("参数IN:"+otherArgs[i]);
				}
			}
		}
		FileOutputFormat.setOutputPath(job,new Path(otherArgs[otherArgs.length-1])); //设置输出路径
		System.out.println("参数OUT："+otherArgs[otherArgs.length-1]);
		FileSystem hdfs=FileSystem.get(conf); //创建⽂件系统
		if(hdfs.exists(new Path(otherArgs[otherArgs.length-1]))){ 
			hdfs.delete(new Path(otherArgs[otherArgs.length-1]),true);
		}
		job.setJarByClass(CourseScoreDriver.class); 
		job.setMapperClass(CourseScoreAverageMapper.class); 
		job.setCombinerClass(CourseScoreAverageReducer.class); 
		job.setReducerClass(CourseScoreAverageReducer.class); 
		job.setOutputKeyClass(Text.class); //设置输出key的类型
		job.setOutputValueClass(IntWritable.class); 
		job.setInputFormatClass(TextInputFormat.class); //设置输⼊格式
		job.setOutputFormatClass(TextOutputFormat.class); //设置输出格式
		System.exit((job.waitForCompletion(true)?0:1));
		System.out.println("Job Finished!");
	}
}

编译运行后打包，上传jar⽂件到Hadoop服务器中并运⾏：

hadoop@hadoop:~$ hadoop jar average.jar hadoop.CourseScoreDriver /data/hadoop /data/output

查看结果，如下图：hadoop@hadoop:~$ hdfs dfs -cat /data/output/part-r-00000

六、结果输出

WordCount实验结果：hadoop@hadoop:~$ hdfs dfs -cat /output/*

考试平均成绩计算结果：

hadoop@hadoop:~$ hdfs dfs -cat /data/output/part-r-00000

亚马逊云科技技术品牌专区

更多推荐

基于STM32的全屋环境智能监测与预警终端设计

亚马逊云科技技术品牌专区

Keras：面向人类的深度学习库

Keras3支持多后端（TensorFlow、PyTorch、JAX），通过设置环境变量切换，兼容指定版本。pip安装后可用Sequential API快速构建CNN模型，如MNIST分类示例，实

亚马逊云科技技术品牌专区

AI行业日报｜2026-09-25：5个热点事件

本期行业动态聚焦于端侧算力重构、自主智能体工作流、开发工具经济模型、网络隔离架构设计与公共部门合规审查。以下结合一线技术场景进行结构化梳理，供 AI 研究员与软件工程团队参考。 PrismML 正将其轻量级语言模型适配至搭载高通处理器的智能眼镜终端 1 。该平台的核心战略指向开源权重模型的本地化运行，旨在最大化挖掘穿戴设备现有算力的利用效率 1 。对于嵌入式与边缘计算开发者而言，此类微型模型向低功