MapReduce基础编程实践（Hadoop 3.5.0和TRAE）

【版权声明】版权所有，请勿转载！

【相关文章推荐】《大数据软件安装和基础编程实践指南》，详细指导VMWare、Ubuntu、Hadoop、HDFS、HBase、Hive、MapReduce、Spark、Flink的安装和基础编程

作者：厦门大学计算机系2025级硕士研究生郑楷阳
指导老师：厦门大学计算机系林子雨副教授

本博客内容与林子雨编著《大数据技术原理与应用（第5版）》（访问教材官网）教材配套。

MapReduce是一种分布式计算编程模型，Hadoop提供了它的开源实现。它把数据处理过程抽象为Map和Reduce两个主要阶段，使编程人员能够通过编写映射和归并逻辑完成并行计算。

本教程以词频统计任务为主线，介绍创建Java项目、添加依赖、编写程序、导出JAR以及读取HDFS上的输入数据进行计算的方法。本次截图所示环境为Ubuntu虚拟机、TRAE CN、Hadoop 3.5.0和JDK 17；JDK运行路径为/usr/lib/jvm/jdk-17.0.20.1+1。

Hadoop需要提前完成安装和HDFS配置。Hadoop 3.5.0服务端要求Java 17，本教程的项目与Hadoop运行环境都使用JDK 17。

本次实际作业日志显示LocalJobRunner，即MapReduce任务在本机执行，输入和输出使用HDFS。第6节将明确采用这一运行方式。

1. 词频统计任务要求

首先在Ubuntu本地文件系统中创建两个文件wordfile1.txt和wordfile2.txt。本实例为了便于观察结果，每个文件只包含两行英文文本。

wordfile1.txt的内容如下：

I love Spark
I love Hadoop

wordfile2.txt的内容如下：

Hadoop is good
Spark is fast

本教程将这两个文件保存到/usr/local/hadoop目录中。可以在Ubuntu终端执行下面的命令创建文件：

cd /usr/local/hadoop
cat > wordfile1.txt <<'EOF'
I love Spark
I love Hadoop
EOF
cat > wordfile2.txt <<'EOF'
Hadoop is good
Spark is fast
EOF

任务要求是：把两个本地文件上传到HDFS的/user/hadoop/input目录，统计该目录下所有输入文件中的单词出现次数，并把结果写入HDFS的/user/hadoop/output目录。

本程序按空白字符分词，区分大小写，不额外去除标点。例如Spark和spark会被视为不同的单词。对于上述输入，预期结果为：

Hadoop  2
I   2
Spark   2
fast    1
good    1
is  2
love    2

单词与次数之间使用制表符分隔。本实例设置一个Reduce任务，输出会按单词的键排序；结果顺序与英文单词出现的先后顺序不同。

开始操作前，先区分下面几类路径：

用途	本教程使用的路径	所属文件系统
原始文本	/usr/local/hadoop/wordfile1.txt和wordfile2.txt	Ubuntu本地文件系统
Java项目	/home/hadoop/Documents/trae_projects/WordCount	Ubuntu本地文件系统，项目位置可自行选择
JAR部署目录	/usr/local/hadoop/myapp	Ubuntu本地文件系统
作业输入	/user/hadoop/input	HDFS
作业输出	/user/hadoop/output	HDFS

2. 在TRAE中创建项目

启动安装在Ubuntu虚拟机中的TRAE CN。如果打开的是图7-1所示页面，点击右上角的“IDE”进入开发界面。

在IDE界面中，可以点击“打开文件夹”，选择准备存放Java项目的父目录。本次截图使用/home/hadoop/Documents/trae_projects；读者也可以选择自己的项目存放目录。

如果尚未安装Java扩展，请在左侧扩展面板中搜索并安装“Extension Pack for Java”。已经完成前面HDFS或HBase实践、并已安装该扩展包的读者，可以直接继续操作。

在左侧资源管理器的空白处点击鼠标右键，选择“创建Java项目...”。也可以按Ctrl+Shift+P打开命令面板，执行“Java: Create Java Project...”。

在项目类型选择界面中选择“No build tools”，创建一个不使用Maven或Gradle的普通Java项目。

随后在文件夹选择窗口中选中项目的父目录，点击右下角的“Select the project location”。这里选择的是父目录，Java扩展会在其中创建项目文件夹。

在项目名称输入框中输入“WordCount”，按Enter确认。

创建完成后，打开新建的WordCount项目文件夹。本次项目目录为/home/hadoop/Documents/trae_projects/WordCount。项目中包含src、bin、lib和.vscode等目录，并在src中生成App.java。

点击打开App.java，等待Java扩展完成项目初始化。如下图所示，底部的状态为“Java: Ready”，左侧下方也已出现“Java Projects”视图。

展开“Java Projects”。如果该视图没有显示，可以在资源管理器标题栏的“...”菜单中勾选“Java Projects”。如果Java扩展提示当前处于轻量模式，按状态栏提示切换到标准模式，才能加载依赖、构建和运行项目。

按Ctrl+Shift+P，执行“Java: Configure Java Runtime”，确认WordCount项目使用JDK 17。本次实际运行截图中的JDK路径为/usr/lib/jvm/jdk-17.0.20.1+1；安装路径不同的读者，应选择自己的JDK 17目录。项目JDK与Java语言服务的运行JDK可以分别设置，不能仅凭语言服务已启动就认定项目JDK已经配置正确。

3. 为项目添加需要用到的JAR包

程序使用Hadoop提供的Java API，因此需要把相应的JAR添加到项目依赖中。本教程的Hadoop安装目录为/usr/local/hadoop。

在“Java Projects”视图中展开WordCount，找到“Referenced Libraries（引用的库）”，点击该行右侧的“＋”，打开JAR文件选择窗口。

先进入/usr/local/hadoop/share/hadoop目录，可以看到common、hdfs、mapreduce和yarn等子目录。图7-8显示的是进入mapreduce之前的目录选择界面。

本实验通过TRAE执行的是编译和无参数的入口检查；实际作业由第6节的Hadoop命令运行。按下面的顺序添加本次截图中的依赖：

添加顺序	目录	选择内容
1	/usr/local/hadoop/share/hadoop/mapreduce	当前目录中的.jar文件，子目录不选
2	/usr/local/hadoop/share/hadoop/common	hadoop-common-3.5.0.jar和hadoop-nfs-3.5.0.jar
3	/usr/local/hadoop/share/hadoop/common/lib	当前目录中的所有.jar文件

先打开mapreduce目录，选中当前目录中的JAR文件，再点击“Select Jar Libraries”。不要把jdiff、lib或sources文件夹作为JAR文件选中

再次点击“Referenced Libraries”右侧的“＋”，进入common目录。按住Ctrl分别选中hadoop-common-3.5.0.jar和hadoop-nfs-3.5.0.jar，再点击“Select Jar Libraries”。

然后重复上述操作，进入common/lib目录，选择其中的所有JAR文件，再点击“Select Jar Libraries”。当该目录下显示的都是JAR文件时，可以使用Ctrl+A全选。这里不能漏掉common/lib中的第三方依赖。

全部添加后，等待Java扩展刷新项目。在“Referenced Libraries”中可以检查已添加的依赖。若后面代码中的org.apache.hadoop导入语句仍显示错误，先核对依赖文件是否选中、项目是否完成加载，以及所选文件是否都来自当前Hadoop 3.5.0安装目录。

4. 编写Java应用程序

在左侧资源管理器中找到WordCount项目的src目录，点击鼠标右键，依次选择“创建Java类”→“类...”。

在输入框中输入“WordCount.java”，按Enter确认。本次截图的输入方式包含.java扩展名，创建后应检查生成的文件名确实为WordCount.java。

打开WordCount.java，清空自动生成的内容，输入下面的完整程序并按Ctrl+S保存。源文件直接放在src目录下，不声明package；public class名称必须与文件名WordCount.java一致。App.java是自动生成的演示文件，后面导出时应选择WordCount作为入口。

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {

    public static class TokenizerMapper
            extends Mapper<Object, Text, Text, IntWritable> {
        private static final IntWritable ONE = new IntWritable(1);
        private final Text word = new Text();

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

    public static class IntSumReducer
            extends Reducer<Text, IntWritable, Text, IntWritable> {
        private final IntWritable result = new IntWritable();

        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> [<in>...] <out>");
            System.exit(2);
        }

        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setNumReduceTasks(1);

        for (int i = 0; i < otherArgs.length - 1; i++) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

5. 编译打包程序

5.1 检查编译与程序入口

保存WordCount.java后，等待Java扩展完成构建。确认“问题”面板中没有阻止编译的错误，再点击编辑器右上角的“Run Java”，或main方法上方的“Run”。

此时没有传入输入和输出路径，程序会在终端显示：

Usage: wordcount <in> [<in>...] <out>

这是参数不足时的用法提示，程序按代码退出，退出状态为2。它说明程序已经进入main方法的参数检查部分，并不表示词频统计作业已经完成。完整运行见第6节。

5.2 导出WordCount.jar

按Ctrl+Shift+P打开命令面板，输入“Java: Export Jar”，选择“Java: 导出到Jar文件...”或对应的“Java: Export Jar...”命令。

在“Export Jar: Determine main class”界面中，选择“WordCount”，按Enter确认。不要选择App，也不要选择“without main class”。

随后进入“Export Jar: Determine elements”界面。取消所有依赖JAR的勾选，只保留bin编译输出，再点击右上角的“确定”。

等待导出结束，确认日志显示导出成功，默认导出到项目根目录下。

在Ubuntu终端中检查JAR内容：

cd ~/Documents/trae_projects/WordCount
jar tf WordCount.jar

输出中应包含下面三个编译类；JAR中还可能包含App.class和清单等其他条目：

WordCount.class
WordCount$TokenizerMapper.class
WordCount$IntSumReducer.class

如果缺少WordCount的主类或内部类，返回项目检查编译错误，完成构建后重新导出。

5.3 复制到Hadoop部署目录

在Ubuntu终端中执行：

mkdir -p /usr/local/hadoop/myapp
cd ~/Documents/trae_projects/WordCount
cp WordCount.jar /usr/local/hadoop/myapp/
cd /usr/local/hadoop/myapp
ls

其中项目目录应与第2节创建项目时的实际选择一致。下图是执行结果。

此时WordCount.jar已经部署到/usr/local/hadoop/myapp，下一步使用Hadoop命令运行它。

6. 运行程序

6.1 启动HDFS并检查输入目录

如果HDFS尚未启动，在Ubuntu终端执行：

cd /usr/local/hadoop
./sbin/start-dfs.sh

服务已启动时无需重复启动。本实验沿用已有Hadoop安装与HDFS配置。可以查看默认文件系统设置：

cd /usr/local/hadoop
./bin/hdfs getconf -confKey fs.defaultFS

结果应指向已配置的HDFS服务，即以hdfs://开头。如果显示file:///，请先完成HDFS配置再继续，这时使用/user/hadoop/input并不会自动把数据放到HDFS中。

创建输入目录并查看其中的内容：

cd /usr/local/hadoop
./bin/hdfs dfs -mkdir -p /user/hadoop/input
./bin/hdfs dfs -ls /user/hadoop/input

第一次完成本实验时，input应为空。如果该目录已用于其他练习，先核对已有文件；可以改用另一个空的HDFS目录，例如/user/hadoop/wordcount/input，并同步修改后面的创建、上传和运行命令。程序会统计输入目录中的所有有效输入文件，额外文本会改变词频结果。

6.2 上传两个输入文件

将第1节创建的两个本地文件上传到HDFS：

cd /usr/local/hadoop
./bin/hdfs dfs -put ./wordfile1.txt ./wordfile2.txt /user/hadoop/input/
./bin/hdfs dfs -ls /user/hadoop/input

检查HDFS中是否有wordfile1.txt和wordfile2.txt，也可以直接核对它们的内容：

cd /usr/local/hadoop
./bin/hdfs dfs -cat /user/hadoop/input/wordfile1.txt
./bin/hdfs dfs -cat /user/hadoop/input/wordfile2.txt

6.3 确认输出目录不存在

MapReduce提交前，/user/hadoop/output必须不存在，不要事先创建这个目录。

第一次运行且尚无output目录时，可以直接进行下一步。如果上次实验已经生成了output，且旧结果无需保留，可以删除本次实验的旧输出：

cd /usr/local/hadoop
./bin/hdfs dfs -rm -r /user/hadoop/output

如果需要保留旧结果，应把下一步命令的输出路径改成一个尚不存在的新目录，例如/user/hadoop/output2。重复运行时无需删除和重建输入目录。

6.4 提交词频统计作业

在Ubuntu终端执行：

cd /usr/local/hadoop
./bin/hadoop jar ./myapp/WordCount.jar input output

WordCount.jar是本地JAR文件，最后两个路径分别为HDFS输入目录和输出目录。

运行过程中会显示任务进度和计数器。下图就是本次的实际运行记录，可以看到map与reduce均达到100%，并显示作业成功完成。

6.5 查看词频统计结果

作业成功后，结果写入HDFS的/user/hadoop/output目录。可以执行下面的命令，只读取Reduce输出文件：

cd /usr/local/hadoop
./bin/hdfs dfs -ls /user/hadoop/output
./bin/hdfs dfs -cat '/user/hadoop/output/*'

预期结果为：

Hadoop  2
I   2
Spark   2
fast    1
good    1
is  2
love    2

实际运行结果如下图所示：

至此，完成了创建Java项目、编写MapReduce词频统计程序、打包部署和查看结果的完整练习。再次运行时，保持输入文件不变，并删除不再需要的旧实验输出，或指定新的输出目录。