4. WordCount实战：需求分析、Mapper实现、Reducer实现、Driver驱动类

好，咱们今天来点真格的。

前面讲了那么多MapReduce的理论，什么分而治之、数据本地性、Shuffle过程……说实话，光听这些容易犯困。我当年学的时候也是这样，直到亲手跑通了一个WordCount，才真正理解了MapReduce到底在干什么。

WordCount就是大数据界的"Hello World"。它简单，但五脏俱全。你把它吃透了，后面再复杂的业务逻辑，无非是在这个框架上做文章。

4.1 需求分析：到底要算什么？

需求其实就一句话：给定一堆文本文件，统计每个单词出现的总次数。

举个例子，输入是：

Hello World
Hello Hadoop
Hello MapReduce

输出应该是：

Hadoop  1
Hello   3
MapReduce   1
World   1

嗯，看起来很简单对吧？但你要想想，如果文件有100个，每个文件有1GB，单机跑不动了怎么办？这时候MapReduce就派上用场了。

核心思路：把"统计单词"这件事拆成两个阶段。

  • Map阶段：每个节点读自己负责的那部分数据，把每行拆成单词，输出 <单词, 1> 这样的键值对。
  • Reduce阶段：把相同单词的"1"全部加起来，得到最终次数。

说白了，Map就是"分"，Reduce就是"合"。这个思想贯穿整个MapReduce，你以后设计任何任务，都可以先问自己：怎么分？怎么合？

4.2 Mapper实现：把数据拆成最小单元

Mapper是MapReduce的第一步。它的输入是原始文本的一行，输出是 <单词, 1> 这样的键值对。

我直接上代码，咱们边看边聊。

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class WordCountMapper 
    extends Mapper<LongWritable, Text, Text, IntWritable> {

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        
        // 拿到当前行数据，转成String
        String line = value.toString();
        
        // 按空格或制表符拆分
        String[] words = line.split("\\s+");
        
        // 遍历每个单词，输出 <单词, 1>
        for (String w : words) {
            if (w.length() > 0) {  // 过滤空字符串
                word.set(w);
                context.write(word, one);
            }
        }
    }
}

这里有几个细节，我当年踩过坑，得跟你说说。

个人经验：

  • 类型别搞错：Mapper的四个泛型参数分别是：输入Key类型、输入Value类型、输出Key类型、输出Value类型。输入Key是行偏移量（LongWritable），输入Value是行内容（Text）。别写成String，Hadoop有自己的序列化类型。
  • 重用对象：你看我声明了word和one两个成员变量，每次map方法里只是set值，而不是new新对象。为什么？因为map方法会被调用几百万次，每次都new对象，GC会把你搞崩溃的。这个优化习惯，我建议你从一开始就养成。
  • 空字符串过滤：split之后可能会有空串，比如两个空格连在一起。加个长度判断，避免输出空单词。

4.3 Reducer实现：把零散的"1"加起来

Mapper输出的是 <单词, 1> 的流。经过Shuffle和Sort之后，Reducer收到的数据变成了 <单词, [1, 1, 1, ...]> 这样的形式。Reducer要做的，就是把这一串"1"加起来。

import java.io.IOException;
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();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, 
            Context context) throws IOException, InterruptedException {
        
        int sum = 0;
        // 遍历所有value，累加
        for (IntWritable val : values) {
            sum += val.get();
        }
        
        result.set(sum);
        context.write(key, result);
    }
}

这段代码看起来简单，但有个地方容易让人困惑：Iterable<IntWritable> values。为什么是Iterable而不是List？

嗯，这里要解释一下。Hadoop为了节省内存，不会把所有value都加载到内存里。它用迭代器的方式，每次只给你一个值，处理完就丢掉。你想想看，如果一个单词出现了几亿次，全放内存里那还得了？

我曾经犯过的错：刚开始写Reducer时，我试图把values转成List来操作，结果跑大文件时直接OOM了。后来才明白，Iterable的设计就是为了流式处理，千万别在reduce方法里缓存所有数据。

4.4 Driver驱动类：把一切串起来

Mapper和Reducer写好了，但怎么让它们跑起来？这就需要Driver类了。它负责配置作业、设置输入输出路径、指定Mapper和Reducer类，然后提交给集群执行。

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCountDriver {

    public static void main(String[] args) throws Exception {
        // 1. 创建配置对象
        Configuration conf = new Configuration();
        
        // 2. 创建Job实例
        Job job = Job.getInstance(conf, "Word Count");
        
        // 3. 设置Jar包主类
        job.setJarByClass(WordCountDriver.class);
        
        // 4. 设置Mapper和Reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        
        // 5. 设置Mapper输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        // 6. 设置最终输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        // 7. 设置输入输出路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        // 8. 提交作业并等待完成
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

这里有几个关键点，我单独拎出来说。

配置项 说明 常见错误
setJarByClass 告诉Hadoop从哪里找Jar包 不设置的话，分布式运行时找不到类
setMapOutputKeyClass Mapper输出的Key类型 如果和Reducer输出不同，必须单独设置
setOutputKeyClass Reducer输出的Key类型 如果和Mapper输出相同，可以省略setMapOutputKeyClass
FileInputFormat 指定输入路径，支持目录 路径不存在会报错
FileOutputFormat 指定输出路径 输出目录不能已存在，否则作业失败

我个人的习惯：在开发阶段，我会把Mapper和Reducer的输出类型都显式设置一遍，哪怕它们一样。这样代码更清晰，别人一看就知道每个阶段的输出是什么。另外，输出路径我通常会加一个时间戳，避免重复运行时报错。

4.5 运行与验证

代码写完了，怎么跑？假设你已经在Hadoop集群上配好了环境，打成Jar包后，用以下命令提交：

hadoop jar wordcount.jar WordCountDriver /input /output_20250101

跑完之后，去输出目录看看结果：

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

如果一切顺利，你会看到类似这样的输出：

Hadoop  1
Hello   3
MapReduce   1
World   1

嗯，到这里，你的第一个MapReduce程序就跑通了。

总结一下：

  • Mapper：负责拆分，输出 <单词, 1>。注意类型和对象重用。
  • Reducer：负责聚合，把相同单词的"1"加起来。注意Iterable的流式特性。
  • Driver：负责组装，配置作业参数，提交运行。注意路径和类型设置。

WordCount虽然简单，但它是理解MapReduce的基石。后面讲Combiner、Partitioner、自定义InputFormat，都是在WordCount的基础上做扩展。所以，我建议你亲手敲一遍代码，跑通它。遇到问题别怕，每个大数据工程师都是从WordCount开始的。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →