💝💝💝首先，欢迎各位来到我的博客，很高兴能够在这里和您见面！希望您在这里不仅可以有所收获，同时也能感受到一份轻松欢乐的氛围，祝你生活愉快！
💝💝💝如有需要请大家订阅我的专栏【大数据系列】哟！我会定期更新相关系列的文章
💝💝💝关注！关注！！请关注！！！请大家关注下博主，您的支持是我不断创作的最大动力！！！

引言

从Google发布MapReduce论文至今已有二十年，它仍然是理解大数据处理绕不开的第一课。很多人觉得MapReduce已经“过时”了——毕竟Spark、Flink跑得更快。但MapReduce带来的思想转变，影响了此后所有的分布式计算框架：把一个大问题拆成许多小问题，分别解决，再汇总结果。

这八个字，就是“分而治之”。

一、分而治之：古老智慧遇上大数据

分而治之（Divide and Conquer）是一个极其古老的算法思想。维基百科上说，分治法就是把一个复杂的问题分成两个或多个相同或相似的子问题，再把子问题分成更小的子问题……直到最后子问题可以简单地直接求解，然后将子问题的解合并成原问题的解。

人类历史上最早的“分治法”例子大概是这样的：要数清一大盒绿豆有多少粒，一个人一粒粒数太慢。叫来100个人，每人分一小把绿豆，各自数清自己手里的粒数，最后把所有人的结果加起来。这就是最原始的MapReduce。

Google 在2004年发表的MapReduce论文中，正是把这个古老思想迁移到了分布式计算场景。论文中给出的定义非常数学化：

MapReduce 是一种编程模型，用于处理和生成大规模数据集。用户指定一个 map 函数，通过它把输入键值对处理成一组中间键值对；再指定一个 reduce 函数，用于将具有相同中间键的所有中间值合并起来。

简单说：Map 负责分拆和初步处理，Reduce 负责汇总和最终计算。

二、Map + Reduce：从单词计数看懂一切

几乎所有的MapReduce教程都会用 WordCount（词频统计） 作为入门案例。因为它足够简单，又能完整展现“分而治之”的全部步骤。

问题：给你几百个GB的文本文件，统计每个单词出现的总次数。

2.1 直面问题：单机好不好做？

一台普通服务器单机处理几百GB的文本，需要把整个文件读一遍（内存装不下，只能流式读），用哈希表统计每个单词的次数。理论上可行，但处理时间可能是几个小时甚至几天。如果文件是TB级别甚至是PB级别，单机就不行了——要么内存装不下哈希表，要么处理时间长得无法接受。

MapReduce的思路是：把这堆文件切成很多块，每台机器处理一小块的计算结果是部分统计结果（Map阶段），然后把所有机器的部分统计结果合并成最终结果（Reduce阶段）。

在这个过程中，我们不需要关心文件有多大——只要集群的规模足够大，总能把处理时间压缩到可接受的范围。

2.2 Map 阶段：各管一摊，先出局部结果

假设文本内容是这样的三行：

hello world
hello hadoop
hello mapreduce

把这三行文本切分成三个独立的输入分片（InputSplit），三个Map任务同时处理。每个Map任务的逻辑完全相同：

• 输入的 key：行号（不重要）
• 输入的 value：一行文本
• 输出：若干个 (word, 1) 键值对

Map 1 处理第一行 "hello world"，输出：

(hello, 1)
(world, 1)

Map 2 处理第二行 "hello hadoop"，输出：

(hello, 1)
(hadoop, 1)

Map 3 处理第三行 "hello mapreduce"，输出：

(hello, 1)
(mapreduce, 1)

至此，Map 阶段完成了“分”的动作——把原始数据拆成小块，各自算出了一份局部结果。

2.3 Shuffle（洗牌）：Map 到 Reduce 的桥梁

Map 阶段产出的中间结果，不能直接送给 Reduce。MapReduce 框架会自动做一件事——把相同 key 的所有 value 归集到一起，再送给 Reduce。

这个过程叫 Shuffle（洗牌），是Hadoop里最复杂、最耗时的环节。

对于上面的例子，shuffle 之后会形成这样的数据结构：

hello -> [1, 1, 1]
world -> [1]
hadoop -> [1]
mapreduce -> [1]

只有 hello 这个 key 需要进入 Reduce 计算（因为它有多个值）。一个 key 只给一个 Reduce 任务，但一个 Reduce 任务可以处理多个 key。

2.4 Reduce 阶段：汇总全局结果

Reduce 任务的输入是：(hello, [1,1,1])。Reduce 的逻辑就是把列表里所有的 1 加起来：

sum = 0
for each v in values:
    sum += v
output (hello, sum)

最终，三个 Reduce 任务会输出：

(hello, 3)
(world, 1)
(hadoop, 1)
(mapreduce, 1)

所有输出合并在一起，就是最终结果。大功告成。

三、不只是思想：MapReduce 的完整运行机制

分而治之是灵魂，但要让它在分布式环境下稳定运行，还需要一整套精心设计的机制。

3.1 从代码到执行

我们写的 MapReduce 程序，在Hadoop中会经历四个层级的变化：

层级	作用
Job	用户提交的完整计算任务
Task	Job 拆解出的 Map Task 或 Reduce Task
Attempt	每个 Task 可以重试多次，每次重试就是一个 Attempt
Record	Task 处理的最小数据单元，通常是文件中的一行

3.2 输入分片：如何决定启动多少个 Map

HDFS 上存储的大文件被切分成 Block（默认128MB）。MapReduce 在处理时，会按照 逻辑上的分片（InputSplit） 来决定启动多少个 Map 任务。通常一个分片对应一个 Block，所以一个128MB的 Block 就会启动一个 Map 任务去处理它。

但也不一定。分片大小是可调的，比如设置分片大小为64MB，那么128MB的文件会被切成两个分片，启动两个 Map 任务。分片数量的多少决定了 Map 任务的并行度。

3.3 中间结果如何处理

Map 的输出不是直接写到 HDFS——那样太慢了。Map 的输出先写到本地磁盘的缓冲区，满了之后溢写到本地文件系统的临时文件。Reduce 再从每个 Map 所在的节点上拉取对应分区的数据。

Shuffle 过程是所有分布式计算框架的难点，因为它涉及大量的网络传输和排序操作。Hadoop 在 1.x 和 2.x 中花了极大的精力优化这个过程。

四、代码实战：从 WordCount 开始

下面是一个完整的 WordCount 示例，使用 Hadoop 新版 API（mapreduce 包，非旧的 mapred 包）。

4.1 Maven 依赖

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>3.3.6</version>
</dependency>

4.2 Mapper 类

package com.example.wordcount;

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

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

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 line = value.toString();
        // 按空格切分单词
        StringTokenizer tokenizer = new StringTokenizer(line);
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            // 输出 (word, 1)
            context.write(word, one);
        }
    }
}

4.3 Reducer 类

package com.example.wordcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

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);
    }
}

4.4 Driver 类（作业配置与提交）

package com.example.wordcount;

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 {
        if (args.length != 2) {
            System.err.println("Usage: WordCountDriver <input path> <output path>");
            System.exit(-1);
        }

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");

        // 设置主类
        job.setJarByClass(WordCountDriver.class);

        // 设置 Mapper 和 Reducer 类
        job.setMapperClass(WordCountMapper.class);
        job.setCombinerClass(WordCountReducer.class);  // 可选，用于本地聚合优化
        job.setReducerClass(WordCountReducer.class);

        // 设置输出 key/value 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置输入输出路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 提交作业并等待完成
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

4.5 打包与运行

# 打包成 jar
mvn clean package

# 上传测试文件到 HDFS
hdfs dfs -mkdir -p /user/hadoop/input
hdfs dfs -put /path/to/your/textfile.txt /user/hadoop/input/

# 运行作业
hadoop jar target/wordcount-1.0-SNAPSHOT.jar com.example.wordcount.WordCountDriver \
    /user/hadoop/input /user/hadoop/output

# 查看结果
hdfs dfs -cat /user/hadoop/output/part-r-00000

五、MapReduce 的“分”与“合”再深入

5.1 Map 端还能做更多

MapReduce 允许在 Map 之后、Reduce 之前加一个 Combiner（本地归约器），本质上就是在 Map 端先对自己的局部结果做一次 Reduce。上例中的 Combiner 直接用了 Reducer 类——因为 WordCount 的合并操作满足交换律和结合律。

用了 Combiner 之后，Map 端输出 (hello, 1), (hello, 1), (hello, 1) 会被先合并成 (hello, 3) 才发给 Reduce，网络传输的数据量大大减少。

5.2 Partitioner：决定哪个 key 去哪个 Reduce

Reduce 默认只有一个，但生产环境不会这样——只有一个 Reduce 能力太差。当 Reduce 数量 > 1 时，就需要一个规则来分配 key 给不同的 Reduce。默认的分配规则是 哈希取模：

partition = hash(key) % numReduceTasks

这样，同一个 key 总是进入同一个 Reduce 分区，才能保证后续计算正确。

5.3 排序是 MapReduce 的内置特性

在数据交给 Reduce 之前，MapReduce 框架会按照 key 对中间结果进行排序。这个排序是自动发生的，不需要用户编写任何排序代码。它带来了两个好处：

1. Reduce 阶段处理时，相同 key 的 value 天然地连续排列在一起
2. Reduce 可以按顺序处理 key，方便实现某些需要排序输出的业务

这也解释了为什么 MapReduce 在排序类的任务中特别擅长（比如全局排序、Top N）。

六、MapReduce 的局限与演进

6.1 为什么有人说 MapReduce“慢”

MapReduce 的设计目标不是快，而是稳——它能处理极大规模的数据，而且容错性极强，适合一次写、多次读的批处理场景。

但它的短板也很明显：

问题	原因
中间结果写磁盘	Map 的输出写到本地磁盘，Reduce 再从磁盘读取，IO 开销大
启动开销大	每个任务都是一个独立的 JVM 进程，启停耗时长
不适合迭代计算	机器学习算法往往需要多轮迭代，每轮都要重启作业
不适合实时计算	从作业提交到结果产出，分钟级延迟是常态

6.2 Spark 的改进

Spark 把中间结果尽量放在内存中，大幅提升了迭代计算的速度。但在很多场景下（如超大数据的 ETL），HDFS 的稳定性和成本优势依然让 MapReduce 有一席之地。

今天的大数据平台往往是混合架构：HDFS 做存储，YARN 或 Kubernetes 做资源调度，Spark / Flink / Hive 做计算。MapReduce 虽然不再像十年前那样耀眼，但它开创的思想已经成为这片土地的基础。

七、总结：一张图回顾 MapReduce 的精髓

核心要点：

• 分而治之：把大任务拆小，并行处理
• 计算向数据移动：把代码分发到数据所在的节点，而不是把数据搬运到代码所在的位置
• 容错性优先：任务失败自动重试，整个作业不会因为单点故障而失败
• 抽象隐藏复杂性：用户只关心 Map 和 Reduce 的业务逻辑，分布式执行的细节由框架处理

MapReduce 不是最快的计算引擎，但它一定是最好的入门教材。理解了它，你就能理解所有分布式计算框架最底层的逻辑。

---

你第一次接触 MapReduce 时，理解“分而治之”花了多长时间？有没有什么让你恍然大悟的瞬间？欢迎评论区分享你的故事～

❤️❤️❤️觉得有用的话点个赞 👍🏻 呗。
❤️❤️❤️本人水平有限，如有纰漏，欢迎各位大佬评论批评指正！😄😄😄
💘💘💘如果觉得这篇文对你有帮助的话，也请给个点赞、收藏下吧，非常感谢!👍 👍 👍
🔥🔥🔥Stay Hungry Stay Foolish 道阻且长,行则将至,让我们一起加油吧！🌙🌙🌙