MapReduce 编程

本篇基于课程实验讲义，介绍如何使用 Eclipse 编写并运行一个 MapReduce 任务——经典的 WordCount 单词计数程序。内容包括开发环境所需的 Hadoop JAR 包、Mapper/Reducer/Driver 的完整代码、导出与手动编译 JAR 的方法，以及在 Hadoop 集群上准备输入数据、提交作业并查看统计结果的全过程。Hadoop 安装目录为 /usr/local/hadoop。

一、编程模型与程序结构

MapReduce 作业（Job）的核心由两部分组成：

• Mapper：对输入的每一行数据进行处理，将其拆分成 <单词, 1> 形式的键值对输出。
• Reducer：对 Mapper 输出的相同键（单词）的所有值进行汇总，累加得到每个单词的出现次数。
• Driver（主程序）：配置并提交作业，指定 Mapper/Reducer 类、输入输出类型与输入输出路径。

在 WordCount 示例中，各泛型参数含义如下：

二、开发环境准备

使用 Eclipse（压缩包 eclipse2018.zip），新建一个 Java 项目，在其中创建类 WordCount.java，包名为 com.testing。

向项目中导入以下路径下的所有 JAR 文件：

/usr/local/hadoop/share/hadoop/common/
/usr/local/hadoop/share/hadoop/common/lib
/usr/local/hadoop/share/hadoop/mapreduce
/usr/local/hadoop/share/hadoop/mapreduce/lib

三、WordCount 完整代码

package com.testing;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.io.IntWritable;

public class WordCount {

    /**
     * Mapper 处理类
     *
     * Object:       输入数据的具体内容
     * Text:         每行的文本数据
     * Text:         每个单词分解后的统计结果（输出键）
     * IntWritable:  输出记录的结果（输出值）
     *
     * @author jim
     */
    @SuppressWarnings("unused")
    private static class WordCountMapper extends Mapper<Object, Text, Text, IntWritable> {
        /**
         * 进行 Map 处理
         */
        @Override
        protected void map(Object key, Text value,
                           Mapper<Object, Text, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            // 获得需要处理的字符串
            String lineContent = value.toString();
            // 按空格进行分割，将单词组成数组
            String result[] = lineContent.split(" ");
            for (int x = 0; x < result.length; x++) {
                // 参数一表示要输出的字符串，参数二表示计数 1
                context.write(new Text(result[x]), new IntWritable(1));
            }
        }
    }

    /**
     * Reducer 处理类
     *
     * Text:         Map 输出的文本内容
     * IntWritable:  Map 处理的个数
     * Text:         Reduce 输出文本
     * IntWritable:  Reduce 的输出个数
     *
     * @author jim
     */
    private static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values,
                              Reducer<Text, IntWritable, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            // 单词出现次数的计数器
            int sum = 0;
            for (IntWritable count : values) {
                sum += count.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

    public static void main(String args[]) throws Exception {
        if (args.length != 2) {
            System.out.println("Usage: hadoop jar filename.jar input_filename.txt output_path");
            System.exit(1);
        }

        Configuration conf = new Configuration();
        // 将命令行参数组装成数组
        String[] argArray = new GenericOptionsParser(conf, args).getRemainingArgs();

        // 定义 Hadoop 作业
        Job job = Job.getInstance();
        job.setJarByClass(WordCount.class);

        // Mapper 设置
        job.setMapperClass(WordCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // Reducer 设置
        job.setReducerClass(WordCountReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 定义路径
        FileInputFormat.addInputPath(job, new Path(argArray[0]));
        FileOutputFormat.setOutputPath(job, new Path(argArray[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

代码说明

• Mapper：读取每行文本，按空格 split(" ") 切分为单词数组，对每个单词输出 <单词, 1>。
• Reducer：遍历同一单词的所有 IntWritable 计数值，累加求和后输出 <单词, 总次数>。
• Driver：
  • 要求命令行传入 2 个参数：输入路径与输出路径；
  • 通过 GenericOptionsParser 解析参数；
  • 设置 Mapper/Reducer 及输出键值类型；
  • FileInputFormat/FileOutputFormat 指定输入输出路径；
  • job.waitForCompletion(true) 提交作业并等待完成。

注：源码中 import java.io.IOException; 重复出现，且注释“安装空格进行过滤”应为“按空格进行过滤”，已在上方修正。

四、打包 JAR

4.1 使用 Eclipse 导出

将项目导出为 JAR 文件 WordCount.jar，导出时需设置主执行类（Main Class）为 com.testing.WordCount。

将 JAR 上传到 Hadoop 虚拟机的 /root/WordCount.jar。

4.2 手动编译（无 Eclipse 时）

如果没有 Eclipse，可按以下步骤手动编译 Java 程序：

export HADOOP_HOME=/usr/local/hadoop
export CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath):/usr/local/jdk/lib/*

cd /root/
javac WordCount.java
jar -cvf WordCount.jar ./WordCount*.class

五、在 Hadoop 上运行

5.1 启动 Hadoop 并验证

start-all.sh        # 启动 Hadoop
jps                 # 验证 Hadoop 是否启动成功

5.2 准备输入数据

在 HDFS 中创建测试目录 /input：

hadoop fs -mkdir /input
hadoop fs -ls /

输出如下，说明目录创建成功：

Found 1 items
drwxr-xr-x   - root supergroup          0 2019-01-01 06:12 /input

在本地建立测试文件并上传到 HDFS：

echo this is my computer this is my computer > /root/input.txt
hadoop fs -put /root/input.txt /input/
hadoop fs -cat /input/input.txt

this is my computer this is my computer

5.3 提交作业

执行 JAR 时需要保证 HDFS 中的输出目录 /output 不存在；若已存在，作业提交阶段就会抛出 org.apache.hadoop.mapred.FileAlreadyExistsException，必须先删除后重跑：

hdfs dfs -rm -r /output    # 如果目录已存在才需要执行
cd /root
hadoop jar WordCount.jar /input/input.txt /output

若导出 JAR 时没有指定主执行类，运行命令中必须补上全限定类名： hadoop jar WordCount.jar com.testing.WordCount /input/input.txt /output

5.4 查看结果

MapReduce 的输出结果位于输出目录下的 part-r-00000 文件：

hadoop fs -cat /output/part-r-00000

computer    2
is  2
my  2
this    2

结果表示单词计数分析成功。

六、运行流程小结

说明：原始讲义以 WordCount 实验为主，未展开描述 shuffle 阶段（分区、排序、合并、归并）的内部细节，故本篇不额外补充，仅按原文整理编程与运行过程。