2. 核心编程模型：Map阶段详解、Reduce阶段详解、Shuffle机制剖析

好，咱们今天来啃硬骨头。MapReduce 的核心，说白了就三块：Map 怎么干、Reduce 怎么收、中间那层 Shuffle 怎么传。我当年刚接触时，觉得 Map 和 Reduce 挺好理解，但一到 Shuffle 就懵了。后来踩了几个坑，才算真正搞明白。

2.1 Map 阶段：数据的分片与映射

Map 阶段，是数据处理的「第一棒」。它的任务很简单：把输入数据拆成一个个键值对，然后交给用户定义的 map 函数处理。

我习惯把 Map 阶段拆成三个子步骤：

  1. 输入分片（Input Split）：HDFS 上的一个大文件，会被逻辑切分成多个 split。每个 split 对应一个 map task。
  2. RecordReader 读取：把 split 解析成 <key, value> 对。比如 TextInputFormat 默认按行读，key 是行偏移量，value 是行内容。
  3. map 函数处理：你写的 map 逻辑，对每个键值对执行一次，输出新的 <key, value>。

关键点：map 的输出不会直接落盘到 HDFS，而是先写入本地磁盘的环形缓冲区。这是性能优化的第一道关口。

举个例子，统计单词频率：

public static class TokenizerMapper
    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);
        }
    }
}

你看，map 函数每读一行，就拆成单词，每个单词输出 <word, 1>。这里有个细节：context.write 并不是直接写到网络，而是先写到缓冲区。嗯，这里要注意，缓冲区大小默认 100MB，可以通过 mapreduce.task.io.sort.mb 调整。

我的经验：我曾经在一个日志分析项目中，map 输出量特别大，默认缓冲区频繁溢写，导致大量磁盘 I/O。后来我把缓冲区调到 200MB，同时启用压缩（mapreduce.map.output.compress=true），性能提升了 30%。

2.2 Reduce 阶段：数据的归并与聚合

Reduce 阶段，是数据处理的「最后一棒」。它的核心任务是：从多个 map task 拉取属于自己分区的数据，然后合并、排序，最后交给 reduce 函数处理。

Reduce 阶段也分三步：

  1. Shuffle（拉取）：从所有已完成 map task 的节点上，拉取属于自己分区的数据。
  2. Merge & Sort（合并与排序）：把拉来的多个小文件合并成一个大文件，并按 key 排序。
  3. reduce 函数处理：对每个 key 及其对应的 value 列表，执行你写的 reduce 逻辑。

还是单词统计的例子：

public static class IntSumReducer
    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;
        for (IntWritable val : values) {
            sum += val.get();
        }
        result.set(sum);
        context.write(key, result);
    }
}

这里有个容易忽略的点：reduce 函数的输入是 Iterable<IntWritable>，而不是一个完整的列表。这意味着你不能在 reduce 里多次遍历 values——它只能被迭代一次。我见过有同事想先遍历一遍算总数，再遍历一遍算平均值，结果发现第二次遍历是空的。嗯，这个坑我踩过。

避坑指南：我曾经在做一个用户行为分析任务时，reduce 端数据倾斜严重。某个 key 的数据量是其他 key 的 100 倍，导致那个 reduce task 跑了 2 小时，其他 task 几分钟就结束了。解决方案是加一层「二次散列」——在 map 端对 key 加随机前缀，reduce 端先做局部聚合，再去掉前缀做全局聚合。

2.3 Shuffle 机制：Map 与 Reduce 之间的桥梁

Shuffle，是 MapReduce 里最复杂、也最影响性能的环节。说白了，它就是把 map 的输出，按 key 分区、排序、传输到 reduce 端的过程。

我把它拆成两个阶段来讲：

2.3.1 Map 端的 Shuffle

Map 端写完数据后，并不是直接发给 reduce。它先做三件事：

  • 分区（Partition）：根据 key 的哈希值，决定每条记录去哪个 reduce。
  • 排序（Sort）：在内存缓冲区里，按 key 进行快速排序。
  • 溢写（Spill）：缓冲区满了，就溢写到磁盘，生成一个临时文件。多个溢写文件最后会合并成一个。

你看，map 端其实做了很多「预处理」。为什么要这么做？因为 reduce 端拉取数据时，希望拿到的是已经排好序、分好区的数据，这样 reduce 端就不用再花大力气排序了。

性能关键点：map 端的溢写次数越少，性能越好。减少溢写的方法有两个：一是增大缓冲区大小，二是启用 map 端压缩。我一般建议把缓冲区设为 200-300MB，压缩用 Snappy 或 LZ4。

2.3.2 Reduce 端的 Shuffle

Reduce 端的工作，可以概括为「拉、合、排」：

  1. 拉取（Fetch）：reduce task 启动后，会启动多个拉取线程，从各个 map task 节点拉取属于自己分区的数据。
  2. 合并（Merge）：拉来的数据是多个小文件，reduce 端会把这些文件合并成一个大文件。合并过程中会做归并排序。
  3. 排序（Sort）：最终得到一个按 key 排序的大文件，然后逐 key 调用 reduce 函数。

这里有个细节：reduce 端拉取数据时，是边拉取边合并的，而不是等所有数据都拉完再合并。这样可以减少内存占用，也加快处理速度。

我的建议：如果你发现 reduce 端拉取数据很慢，可以调整 mapreduce.reduce.shuffle.parallelcopies 参数，默认是 5，可以适当调大到 10-20。但别太大，否则会压垮网络。我曾经在一个 200 节点的集群上，把这个参数调到 30，结果网络被打满，反而更慢了。

2.4 总结：一张表看懂三个阶段

阶段 核心任务 关键参数 常见问题
Map 读取分片，执行 map 函数，输出键值对 mapreduce.task.io.sort.mb 缓冲区溢写频繁
Shuffle 分区、排序、传输、合并 mapreduce.reduce.shuffle.parallelcopies 数据倾斜、网络瓶颈
Reduce 拉取数据，合并排序，执行 reduce 函数 mapreduce.reduce.memory.mb 内存不足、迭代器陷阱

好了，这一章的内容就到这里。Map 阶段是起点，Reduce 阶段是终点，Shuffle 是连接它们的「高速公路」。你想想看，如果高速公路堵车了，起点和终点再快也没用。所以，性能优化的重点，往往就在 Shuffle 上。下一章，我会专门讲怎么调优 Shuffle，咱们到时候细聊。

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