Hadoop MapReduce 详解

想象一个巨大的文档分类任务，一个人处理要一个月。MapReduce 把这个任务分配给一百个人，每人处理一小部分，最后把结果汇总，半小时就完成了！这就是 MapReduce 的威力！

📑 目录

1. MapReduce 概述
2. 名词解释（命令与术语）
3. Map 阶段详解
4. Shuffle 阶段详解
5. Reduce 阶段详解
6. 完整 WordCount 示例
7. 高级特性
8. MapReduce 与 YARN
9. 监控与调试
10. 性能优化
11. 总结
12. 官方文档与参考

🎯 MapReduce 概述

什么是 MapReduce？

MapReduce 是一种分布式计算框架，用于处理和生成大数据集。它将复杂的并行计算过程抽象为两个阶段：Map（映射）和 Reduce（归约）。

官方定义简述（来源：Apache Hadoop MapReduce）：MapReduce 是建立在 YARN 之上的并行处理大数据集的编程模型。用户实现 Mapper 和 Reducer，框架负责把输入切分为分片、调度 Task、处理 Shuffle 与容错，适合批处理场景。

为什么需要 MapReduce？

想象一个图书馆要统计所有书中的单词出现次数：

• 单机处理 = 一个图书管理员一本本翻，可能要几个月
• MapReduce = 一百个管理员每人负责几本书，最后汇总结果，几天就完成

📂 大数据集

📤 自动分发

🗺️ Map 并行处理

🔀 Shuffle 数据交换

📊 Reduce 汇总结果

📁 最终输出

MapReduce 的核心思想

MapReduce vs 传统计算

MapReduce

📂 读取数据

🗺️ Map 并行

🔀 Shuffle 排序

📊 Reduce 汇总

📁 输出结果

传统计算

📂 读取数据

🖥️ 单机处理

📁 输出结果

Map 与 Reduce 对比（相近概念辨析）

📖 名词解释（命令与术语）

以下对文档中出现的命令、概念、配置项做简要解释，并配上生活例子与「为什么」，便于记忆与理解。

常用命令

核心概念

配置参数（为什么这样设？）

🗺️ Map 阶段详解

Map 的作用

Map 阶段负责将输入数据拆分成独立的键值对，并进行初步处理。

生活类比：

• 公司年终报告
• Map = 各部门各自统计自己的数据
• Reduce = 汇总各部门数据形成总报告

Map 工作流程

📂 输入数据
InputSplit

📖 RecordReader
读取记录

✏️ Map 函数
用户自定义

🔀 Partition
分区

📋 Sort
排序

💾 Spill
溢写磁盘

📦 Merge
合并文件

Map 阶段详细说明

1. InputSplit（输入分片）

📂 大文件 1TB

📄 分片1: 128MB

📄 分片2: 128MB

📄 分片3: 128MB

📄 ...共 8000+ 分片

InputSplit 是 MapReduce 对输入数据的逻辑划分，每个分片由一个 Map Task 处理。

为什么默认 128MB？ 与 HDFS 块大小（默认 128MB）对齐，这样多数分片可以数据本地化（Task 和块在同一节点），减少网络拉取。分片太大则单 Task 耗时长、失败重试成本高；太小则 Task 数过多、调度与启动开销大。

2. RecordReader（记录读取器）

将输入分片转换为键值对记录。为什么键是偏移量（如 LongWritable）？ 文本按行切时，框架需要唯一标识每一行，用「文件内字节偏移」可以保证不重复且便于定位；若不需要 key，Map 里可以忽略不用。

1// TextInputFormat 示例
2// 输入文件：
3// Hello World
4// Hello Hadoop
5
6// 输出键值对：
7// (0, "Hello World")    <- 键是行首偏移量，值是行内容
8// (12, "Hello Hadoop")  <- 键是第二行的偏移量
9

3. Map 函数

用户自定义的核心处理逻辑。

1// WordCount 的 Map 函数
2public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
3
4    private final static IntWritable one = new IntWritable(1);
5    private Text word = new Text();
6
7    @Override
8    protected void map(LongWritable key, Text value, Context context)
9            throws IOException, InterruptedException {
10
11        // 输入：行偏移量, 行内容
12        // "Hello World Hello"
13
14        String line = value.toString();
15        String[] words = line.split("\\s+");
16
17        for (String w : words) {
18            word.set(w);
19            // 输出：(Hello, 1), (World, 1), (Hello, 1)
20            context.write(word, one);
21        }
22    }
23}
24

Map 输出示例：

1输入: "Hello World Hello"
2输出: (Hello, 1)
3      (World, 1)
4      (Hello, 1)
5

4. Partition（分区）

决定每个键值对发送到哪个 Reducer。为什么必须按 key 分区？ 因为 Reduce 是按 key 分组汇总的，同一 key 的所有 value 必须送到同一个 Reducer，否则结果会错（例如 WordCount 里 “Hello” 被拆到两个 Reduce 就得不到正确总数）。

1// 默认的 HashPartitioner
2public class HashPartitioner<K, V> extends Partitioner<K, V> {
3
4    @Override
5    public int getPartition(K key, V value, int numReduceTasks) {
6        // 根据键的哈希值决定分区
7        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
8    }
9}
10

5. Sort（排序）

在每个分区内按键排序。

1排序前：
2(Apple, 1)
3(Banana, 1)
4(Apple, 1)
5(Cherry, 1)
6(Banana, 1)
7
8排序后：
9(Apple, 1)
10(Apple, 1)
11(Banana, 1)
12(Banana, 1)
13(Cherry, 1)
14

6. Spill（溢写）

当内存缓冲区满时（默认 100MB），数据溢写到磁盘。为什么是 80% 就溢写？ 留出 20% 空间给新来的数据，避免写盘期间缓冲区被撑爆；溢写与 Map 输出可并行，提高吞吐。

达到 80%

后台线程

🔄 环形缓冲区
100MB

💾 溢写到磁盘

📋 排序+分区

📁 生成溢写文件

7. Merge（合并）

将多个溢写文件合并成一个大文件。为什么先 Spill 再 Merge？ 因为 Map 输出可能很大，无法一次性在内存排序；先按 80% 阈值一段段排序后写盘，最后多路归并成一个大有序文件，Reduce 拉取时只需顺序读。

🔀 Shuffle 阶段详解

什么是 Shuffle？

Shuffle 是 MapReduce 的核心，负责将 Map 的输出传输到 Reduce。它是 Map 和 Reduce 之间的"数据搬运工"。

为什么叫 Shuffle？
就像洗牌一样，把数据重新排列组合。

Shuffle 工作流程

Reduce端

数据传输

Map端

🗺️ Map 输出

💾 内存缓冲区

🔀 分区排序

💾 溢写磁盘

📦 合并文件

📤 HTTP 拉取

🌐 网络传输

💾 内存缓冲

📋 归并排序

📊 Reduce 输入

Map 端 Shuffle

1// Map 端 Shuffle 配置优化
2// 1. 环形缓冲区大小（默认 100MB）
3conf.set("io.sort.mb", "200");
4
5// 2. 溢写阈值（默认 80%）
6conf.set("io.sort.record.percent", "0.9");
7
8// 3. 溢写工作线程数
9conf.set("io.sort.factor", "10");
10
11// 4. 压缩 Map 输出
12conf.set("mapreduce.map.output.compress", "true");
13conf.set("mapreduce.map.output.compress.codec",
14    "org.apache.hadoop.io.compress.SnappyCodec");
15

Reduce 端 Shuffle

为什么 Reduce 要主动拉取（Pull）而不是 Map 主动推（Push）？ Pull 模式下 Reduce 按自己进度拉取，避免 Map 输出爆满时拖垮 Map 端；且多个 Reduce 可并行从多个 Map 拉，负载更均衡。

🗺️ Map Task 3 🗺️ Map Task 2 🗺️ Map Task 1 📊 Reducer 🗺️ Map Task 3 🗺️ Map Task 2 🗺️ Map Task 1 📊 Reducer 启动拷贝线程 启动拷贝线程 启动拷贝线程 输出文件 1 输出文件 2 输出文件 3 💾 磁盘合并 📋 归并排序 ✏️ 送给 Reduce 函数

Shuffle 性能优化

📊 Reduce 阶段详解

Reduce 的作用

Reduce 阶段负责接收 Map 的输出，按键分组，进行最终汇总计算。

生活类比：

• 各省（Map）统计完人口后
• 中央统计局（Reduce）汇总成全国总人口

Reduce 工作流程

🔀 Shuffle 输出
已排序的键值对

📂 Grouping
按键分组

✏️ Reduce 函数
用户自定义

📁 OutputFormat
写入文件系统

Reduce 函数示例

1// WordCount 的 Reduce 函数
2public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
3
4    private IntWritable result = new IntWritable();
5
6    @Override
7    protected void reduce(Text key, Iterable<IntWritable> values, Context context)
8            throws IOException, InterruptedException {
9
10        // 输入：(Hello, [1, 1, 1, 1, ...])
11
12        int sum = 0;
13        for (IntWritable val : values) {
14            sum += val.get();
15        }
16
17        result.set(sum);
18        // 输出：(Hello, 1000)
19        context.write(key, result);
20    }
21}
22

Reduce 输入结构

1输入格式：键 + 值迭代器
2
3示例：
4键："Hello"
5值：[1, 1, 1, 1, 1, 1, 1, 1, 1, 1]  (10个1)
6
7处理逻辑：
8遍历所有值，求和
9输出：(Hello, 10)
10

Reduce 任务数量

1# 设置 Reduce 任务数量
2conf.setNumReduceTasks(3);
3
4# 如何确定 Reduce 数量？
5# - 太少：单个 Reduce 负担重，成为瓶颈
6# - 太多：小文件增多，集群开销大
7# - 经验公式：0.95 * 集群节点数 * 每节点任务槽位数
8

为什么 0 个 Reduce 也可以？ 有些作业只需要做 Map（如过滤、格式转换），不需要按 key 汇总，设 setNumReduceTasks(0) 即可，输出直接写 HDFS，文件数为 Map 数。

🚀 完整 WordCount 示例

完整代码

1import java.io.IOException;
2import org.apache.hadoop.conf.Configuration;
3import org.apache.hadoop.fs.Path;
4import org.apache.hadoop.io.IntWritable;
5import org.apache.hadoop.io.LongWritable;
6import org.apache.hadoop.io.Text;
7import org.apache.hadoop.mapreduce.Job;
8import org.apache.hadoop.mapreduce.Mapper;
9import org.apache.hadoop.mapreduce.Reducer;
10import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12
13public class WordCount {
14
15    // Mapper 类
16    public static class TokenizerMapper
17            extends Mapper<LongWritable, Text, Text, IntWritable>{
18
19        private final static IntWritable one = new IntWritable(1);
20        private Text word = new Text();
21
22        @Override
23        public void map(LongWritable key, Text value, Context context)
24                throws IOException, InterruptedException {
25
26            // 输入：(行偏移量, 行内容)
27            String line = value.toString();
28            String[] tokens = line.split("\\s+");
29
30            for (String token : tokens) {
31                if (token.length() > 0) {
32                    word.set(token);
33                    context.write(word, one);
34                }
35            }
36        }
37    }
38
39    // Reducer 类
40    public static class IntSumReducer
41            extends Reducer<Text,IntWritable,Text,IntWritable> {
42
43        private IntWritable result = new IntWritable();
44
45        @Override
46        public void reduce(Text key, Iterable<IntWritable> values,
47                           Context context) throws IOException, InterruptedException {
48
49            int sum = 0;
50            for (IntWritable val : values) {
51                sum += val.get();
52            }
53            result.set(sum);
54            context.write(key, result);
55        }
56    }
57
58    // Driver 类
59    public static void main(String[] args) throws Exception {
60
61        // 创建配置
62        Configuration conf = new Configuration();
63
64        // 创建 Job
65        Job job = Job.getInstance(conf, "word count");
66        job.setJarByClass(WordCount.class);
67
68        // 设置 Mapper 和 Reducer
69        job.setMapperClass(TokenizerMapper.class);
70        job.setReducerClass(IntSumReducer.class);
71
72        // 设置输出类型
73        job.setOutputKeyClass(Text.class);
74        job.setOutputValueClass(IntWritable.class);
75
76        // 设置输入输出路径
77        FileInputFormat.addInputPath(job, new Path(args[0]));
78        FileOutputFormat.setOutputPath(job, new Path(args[1]));
79
80        // 等待任务完成
81        System.exit(job.waitForCompletion(true) ? 0 : 1);
82    }
83}
84

编译和运行

1# 1. 编译
2hadoop com.sun.tools.javac.Main WordCount.java
3jar cf wc.jar WordCount*.class
4
5# 2. 创建测试数据
6hdfs dfs -mkdir /user/input
7echo "Hello World Hello Hadoop" > /tmp/test.txt
8hdfs dfs -put /tmp/test.txt /user/input/
9
10# 3. 运行
11hadoop jar wc.jar WordCount /user/input /user/output
12
13# 4. 查看结果
14hdfs dfs -cat /user/output/part-r-00000
15# 输出：
16# Hadoop  1
17# Hello   2
18# World   1
19

🔧 高级特性

Combiner（规约器）

什么是 Combiner？
Combiner 是 Map 端的"小 Reduce"，在 Shuffle 前先进行局部汇总。

是

否

🗺️ Map 输出

有 Combiner?

📊 局部汇总
减少网络传输

🔀 直接 Shuffle

生活类比：

• 没有 Combiner = 每个人把自己统计的纸条全部送到总部
• 有 Combiner = 每个人先汇总自己的结果，只送汇总后的数据

1// 设置 Combiner（与 Reducer 相同）
2job.setCombinerClass(IntSumReducer.class);
3
4// WordCount 示例：
5// Map 输出：(Hello, 1), (Hello, 1), (Hello, 1), (Hello, 1), (Hello, 1)
6// 有 Combiner：(Hello, 5)  <- 只传输 5 而不是 5 个 1
7// 没有 Combiner：传输 (Hello, 1) 五次
8

为什么平均值不能直接用 Combiner？ 因为 (sum1/count1 + sum2/count2)/2 ≠ (sum1+sum2)/(count1+count2) 的等价形式不能简单「先平均再平均」；若用 Combiner，应输出 (局部 sum, 局部 count)，在 Reduce 里再总 sum / 总 count。

数据本地化（Data Locality）

为什么需要数据本地化？
网络传输比磁盘读取慢很多，让计算靠近数据可以显著提升性能。

理想

次优

最差

📂 数据在 Node A

任务调度

✅ Task 在 Node A 执行
DATA_LOCAL

🟡 Task 在同机架执行
RACK_LOCAL

❌ Task 在远程执行
OFF_SWITCH

推测执行（Speculative Execution）

什么是推测执行？
当一个任务运行明显慢于其他任务时，Hadoop 会在另一个节点上启动一个相同任务的备份，哪个先完成就采用哪个的结果。

1# 启用推测执行
2<property>
3    <name>mapreduce.map.speculative</name>
4    <value>true</value>
5</property>
6
7<property>
8    <name>mapreduce.reduce.speculative</name>
9    <value>true</value>
10</property>
11

为什么任务必须幂等？ 因为同一份数据可能被原始 Task 和备份 Task 各写一次（或先写后杀），若写库、发消息等有副作用，就会重复；只读或「同一 key 多次写同一结果」则安全。

自定义 Partitioner

1// 自分区示例：按首字母分区
2public static class FirstLetterPartitioner
3        extends Partitioner<Text, IntWritable> {
4
5    @Override
6    public int getPartition(Text key, IntWritable value, int numPartitions) {
7        String word = key.toString();
8        char firstChar = word.charAt(0);
9
10        // A-M 到 partition 0
11        if (firstChar >= 'A' && firstChar <= 'M') {
12            return 0;
13        }
14        // N-Z 到 partition 1
15        else {
16            return 1 % numPartitions;
17        }
18    }
19}
20
21// 设置自定义 Partitioner
22job.setPartitionerClass(FirstLetterPartitioner.class);
23job.setNumReduceTasks(2);
24

📊 MapReduce 与 YARN

YARN 架构

YARN

提交 Job

👤 Client

🏢 ResourceManager

📅 Scheduler
资源调度

📋 ApplicationsManager
应用管理

🖥️ NodeManager 1

🖥️ NodeManager 2

🖥️ NodeManager 3

📌 ApplicationMaster
每个 Job 一个

📦 Container 1
Map Task

📦 Container 2
Reduce Task

YARN 组件说明

🔍 监控与调试

常用监控命令

1# 1. 查看运行中的任务
2mapred job -list
3
4# 2. 查看任务详情
5mapred job -status job_1234567890000_0001
6
7# 3. 杀死任务
8mapred job -kill job_1234567890000_0001
9
10# 4. 查看任务日志
11mapred job -logs job_1234567890000_0001
12
13# 5. 查看任务计数器
14mapred job -counter job_1234567890000_0001
15

Web UI

1# 访问 MapReduce Web UI
2http://resourcemanager:8088/
3
4# 查看的信息：
5# - 运行中的任务
6# - 历史任务
7# - 任务统计信息
8# - 日志查看
9

常见计数器

🚨 性能优化

优化策略

🚀 MapReduce 性能优化

🔀 减少 Shuffle

⚖️ 优化数据倾斜

⚙️ 调整参数

📊 合理分区

📊 使用 Combiner

🗜️ 压缩中间数据

📉 减少 Map 输出

🔀 自定义 Partitioner

📌 预处理倾斜 Key

💾 调整缓冲区大小

📈 调整并行度

📊 合理设置 Reduce 数

📁 避免小文件

优化配置

1<!-- mapred-site.xml -->
2
3<!-- 启用压缩 -->
4<property>
5    <name>mapreduce.map.output.compress</name>
6    <value>true</value>
7</property>
8
9<!-- Compressor codec -->
10<property>
11    <name>mapreduce.map.output.compress.codec</name>
12    <value>org.apache.hadoop.io.compress.SnappyCodec</value>
13</property>
14
15<!-- 增大 Shuffle 缓冲区 -->
16<property>
17    <name>mapreduce.reduce.shuffle.input.buffer.percent</name>
18    <value>0.7</value>
19</property>
20
21<!-- 并行拷贝数 -->
22<property>
23    <name>mapreduce.reduce.shuffle.parallelcopies</name>
24    <value>10</value>
25</property>
26
27<!-- 推测执行 -->
28<property>
29    <name>mapreduce.map.speculative</name>
30    <value>true</value>
31</property>
32

🎯 总结

核心要点记忆口诀

1MapReduce 大数据
2Map 处理分而治
3Shuffle 传数据排序
4Reduce 汇总出结果
5
6Combiner 减传输
7本地化提性能
8推测执行防慢节点
9YARN 管理调度灵
10

WordCount 完整流程

📂 输入文件

📄 Split 1

📄 Split 2

🗺️ Map 1

🗺️ Map 2

📊 Combiner

🔀 Shuffle & Sort

📊 Reduce

📁 输出文件

生活类比总结

• Map = 各部门统计数据
• Shuffle = 把各部门数据汇总整理
• Reduce = 总部形成最终报告
• Combiner = 各部门先汇总再上报
• 数据本地化 = 就地处理，不搬砖
• 推测执行 = 慢了就叫人帮忙一起做

最后提醒：MapReduce 适合批处理，不适合低延迟交互式查询！对于实时处理，考虑使用 Spark 或 Flink！

📚 官方文档与参考

MapReduce 与其它计算框架对比（相近方案）

《Hadoop MapReduce 详解》 是转载文章，点击查看原文。