4、MapReduce编程：WordCount案例、序列化与分区、Combiner与Shuffle机制

好，咱们今天来啃一块硬骨头——MapReduce。说实话，这玩意儿在十年前可是大数据界的顶流，现在虽然被Spark、Flink抢了不少风头，但它的思想依然是离线批处理的基石。你想想看，Hive、Spark SQL这些高级工具，底层跑的还是MapReduce那套逻辑。所以，这一章咱们不仅要会写WordCount，更要搞懂它背后的序列化、分区、Combiner和Shuffle这些核心机制。

4.1 从WordCount说起：第一个MapReduce程序

我刚开始学MapReduce的时候，第一个练手的案例就是WordCount。说白了，它就是统计一堆文本里每个单词出现的次数。这个例子虽然简单，但麻雀虽小五脏俱全，Map、Shuffle、Reduce三个阶段全齐了。

咱们先看代码，我习惯把Mapper和Reducer分开写，这样结构清晰。

// Mapper类
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    
    private Text word = new Text();
    private final static IntWritable one = new IntWritable(1);
    
    @Override
    protected void map(LongWritable key, Text value, Context context) 
            throws IOException, InterruptedException {
        // 拿到每一行文本
        String line = value.toString();
        // 按空格切分
        String[] words = line.split(" ");
        // 输出 <单词, 1>
        for (String w : words) {
            word.set(w);
            context.write(word, one);
        }
    }
}

// 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;
        // 累加同一个单词的所有计数
        for (IntWritable val : values) {
            sum += val.get();
        }
        result.set(sum);
        context.write(key, result);
    }
}

嗯，这里要注意：Mapper的输入key是行偏移量，value是行内容。输出是<单词, 1>。Reducer收到的是同一个单词的所有1，然后累加。逻辑很简单，对吧？

4.2 序列化：为什么不用Java的Serializable？

你可能会问：为什么MapReduce里用的都是Text、IntWritable这些玩意儿，而不是直接用Java的String、Integer？

原因很简单：性能。Java的Serializable序列化机制太重了，它会保存类的完整信息，包括继承关系、字段名等等。而Hadoop自己搞了一套Writable接口，只保存数据本身，不保存元数据。你想想看，在大数据场景下，动辄上亿条记录，序列化效率差一点，整个作业可能就多跑半小时。

我在项目中遇到过一个问题：有个同事把自定义对象直接用了Java的Serializable，结果MapReduce跑起来慢得离谱。后来改成实现Writable接口，速度提升了将近3倍。所以，记住这个教训：在Hadoop生态里，能用Writable就别用Serializable。

核心要点：Writable接口只序列化数据，不序列化类型信息。体积小、速度快。

4.3 分区：数据是怎么分到不同Reducer的？

Map阶段结束后，数据要分发给不同的Reducer。这个分发规则就是分区（Partition）。默认情况下，Hadoop用HashPartitioner，它根据key的哈希值对Reducer数量取模。

公式很简单：partition = hash(key) % numReduceTasks。

但有时候默认分区不够用。比如，你有一个日志分析任务，想把同一台服务器的日志分到同一个Reducer里。这时候就需要自定义分区器。

// 自定义分区器：按IP地址的前三段分区
public class IPPartitioner extends Partitioner<Text, IntWritable> {
    
    @Override
    public int getPartition(Text key, IntWritable value, int numPartitions) {
        String ip = key.toString();
        String prefix = ip.substring(0, ip.lastIndexOf("."));
        return Math.abs(prefix.hashCode()) % numPartitions;
    }
}

我曾经在一个流量分析项目里用过这个技巧。当时数据量特别大，默认分区导致某些Reducer负载不均，有的跑完了有的还在跑。加了自定义分区后，整个作业的倾斜问题就解决了。

小技巧：自定义分区时，一定要保证getPartition方法返回的值在0到numPartitions-1之间。否则会报IllegalArgumentException。

4.4 Combiner：在Map端先做一次局部聚合

Combiner是个好东西，但很多人容易忽略它。说白了，Combiner就是在Map端先做一次“迷你Reduce”。

拿WordCount来说，Mapper输出的是<hello, 1>、<hello, 1>、<hello, 1>……如果直接把这些数据通过网络传到Reducer，网络开销会很大。Combiner可以在Map端先把这些1累加起来，变成<hello, 3>，再传给Reducer。

代码怎么写？其实Combiner就是Reducer本身。因为WordCount的Reducer逻辑是累加，Combiner也是累加，所以可以直接复用。

// 设置Combiner
job.setCombinerClass(WordCountReducer.class);

嗯，就这么一行。但要注意：Combiner不是万能的。它只适用于满足交换律和结合律的操作，比如求和、计数、最大值。如果是求平均值，就不能直接用Combiner，因为局部平均和全局平均不是一回事。

避坑指南：我曾经在一个求平均值的任务里直接用了Combiner，结果算出来的结果全是错的。后来才意识到，Combiner只能用于局部聚合，不能改变最终结果的语义。求平均值需要自己写一个特殊的Combiner，输出<key, (sum, count)>，然后在Reducer里再算最终平均值。

4.5 Shuffle机制：MapReduce的核心枢纽

Shuffle，说白了就是Map端输出到Reduce端输入之间的数据搬运过程。这个过程最复杂，也最影响性能。我把它拆成两个阶段来讲。

4.5.1 Map端的Shuffle

Mapper输出的数据不会直接写到磁盘，而是先写到一个内存缓冲区（默认100MB）。当缓冲区达到阈值（默认80%）时，会启动一个后台线程，把数据溢写到磁盘。在溢写之前，会做两件事：

  1. 分区：按照Partitioner决定每条数据去哪个Reducer。
  2. 排序：在每个分区内，按key进行排序。

如果设置了Combiner，还会在溢写之前做一次局部聚合。溢写会产生多个小文件，最后Map任务结束时，这些小文件会被合并成一个有序的大文件。

4.5.2 Reduce端的Shuffle

Reduce端会启动多个拷贝线程，从各个Map任务拉取属于自己分区的数据。这些数据先放在内存缓冲区，如果内存不够就写到磁盘。全部拉取完成后，Reduce端会对数据进行一次合并排序，然后交给Reducer处理。

你想想看，这个过程涉及大量的磁盘I/O和网络传输。所以，MapReduce作业慢，十有八九是Shuffle拖了后腿。

优化建议：

  • 增大Map端缓冲区大小（mapreduce.task.io.sort.mb），减少溢写次数。
  • 适当提高Combiner的使用率，减少网络传输量。
  • 调整Reduce端拷贝线程数（mapreduce.reduce.shuffle.parallelcopies），默认是5，可以适当调大。

4.6 实战：完整的WordCount作业配置

最后，咱们把整个作业串起来。我习惯在Driver类里配置所有参数，这样一目了然。

public class WordCountDriver {
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Word Count");
        
        // 设置Jar包
        job.setJarByClass(WordCountDriver.class);
        
        // 设置Mapper和Reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        
        // 设置Combiner（可选）
        job.setCombinerClass(WordCountReducer.class);
        
        // 设置输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        // 设置输入输出路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        // 设置Reducer数量
        job.setNumReduceTasks(2);
        
        // 提交作业
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

运行命令也很简单：

hadoop jar wordcount.jar WordCountDriver /input /output

好了，这一章的内容就到这里。总结一下：WordCount是入门，序列化是基础，分区控制数据流向，Combiner优化性能，Shuffle是核心机制。下一章咱们会深入MapReduce的进阶特性，包括InputFormat、OutputFormat以及如何应对数据倾斜。到时候见。

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