MapReduce 分析

主要计算阶段

1. 输入分片（Input Splitting），每个 Split 由一个 Mapper 任务处理。
2. Map 阶段（Mapper Tasks 并行执行），每个 Mapper 执行以下子流程：
   1. 读取输入（RecordReader）：将字节流解析为 <key, value> 对
   2. 执行用户定义的map()函数
   3. 缓存与排序（Map 输出缓冲区）
      1. map()输出写入内存环形缓冲区（默认 100MB）。
      2. 当缓冲区达到阈值（默认 80%），触发 spill（溢写） 到本地磁盘。
      3. Spill 前对数据按 key 排序（快速排序）。
      4. 可选：应用 Combiner（本地 Reduce，减少网络传输）。
   4. Merge 多个 Spill 文件：将所有 spill 文件 合并成一个有序文件（每个 Reducer 对应一个分区）。
   5. 📌 此时 Map 输出已按 Reducer ID 分区，并在每个分区内按键排序。
3. Shuffle 阶段（数据从 Map → Reduce）：最耗时、最关键的阶段，完全由框架自动完成：
   1. Partitioning（分区）：Map 输出通过 Partitioner（默认 HashPartitioner）决定发送给哪个 Reducer
   2. Copy（拉取数据）：Reducer 启动后，通过 HTTP 从各个 Mapper 所在节点 拉取属于自己分区的数据
   3. Merge & Sort（归并排序）：Reducer 将来自多个 Mapper 的有序数据块进行多路归并排序，形成全局有序的输入流。
   4. 🔹 Shuffle 结束时，Reducer 的输入是：<font style="background-color:#E8F7CF;"><key, [v1, v2, ..., vn]></font>，且 key 全局有序。
4. Reduce 阶段（Reducer Tasks 并行执行），每个 Reducer 执行：
   1. 执行用户定义的reduce()函数

• 框架保证：同一个 key 的所有 value 一次性传给一个 reduce() 调用。
• 输出结果：每个 Reducer 生成一个输出文件

数据流图解（简化版）

1. HDFS Input
2. [InputFormat]→ 分片（Splits）
3. [Mapper]→ (map()→ buffer → sort → spill → merge)
4. Local Disk (Map Output, partitioned & sorted)
5. Shuffle (Copy + Merge + Sort)
6. [Reducer]→ (reduce()→ HDFS Output)
7. HDFS Output (part-r-xxxxx)

MapReduce 易出现性能瓶颈的阶段

Shuffle 阶段（最常见瓶颈）

原因：网络 I/O、数据倾斜

1. 数据需通过网络从所有 Mapper 传输到 Reducer（I/O + 网络开销大）。
2. 若数据倾斜（某些 key 数据量极大），会导致个别 Reducer 负载过高（“拖后腿”）。
3. 默认使用 HTTP 传输，效率低于现代 RPC 协议

表现：Job 卡在 “map 100% reduce 0%” 长时间不动。

优化手段：

1. 启用压缩中间数据：

mapreduce.map.output.compress=true mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.SnappyCodec

2. 调整 Reducer 数量（避免太少或太多）。
3. 使用 Combiner 减少 shuffle 数据量。
4. 处理 数据倾斜（如加盐、skew join）。

Map 阶段的 Spill 和 Merge

原因：频繁磁盘写入

1. 内存缓冲区不足导致频繁 spill 到磁盘（大量磁盘 I/O）。
2. 多次 spill 文件合并时 CPU 和磁盘压力大。

优化：

1. 增大<font style="color:rgb(6, 10, 38);">mapreduce.task.io.sort.mb</font>（如 512MB）。
2. 增大<font style="color:rgb(6, 10, 38);">mapreduce.map.sort.spill.percent</font>（如 0.9）减少 spill 次数。
3. 使用更快的磁盘（SSD）或增加内存。

Reduce 阶段的 Merge 和内存管理

原因：内存不足导致磁盘 merge

1. Reducer 需要同时处理来自数百/数千个 Mapper 的数据块，内存压力大。
2. 若无法全部加载到内存，会 spill 到磁盘再 merge，造成额外 I/O。

相关参数：

1. <font style="color:rgb(6, 10, 38);">mapreduce.reduce.shuffle.input.buffer.percent</font>：控制用于缓存 shuffle 数据的堆内存比例。
2. <font style="color:rgb(6, 10, 38);">mapreduce.reduce.merge.inmem.threshold</font>：内存中最多合并多少个 map 输出。

优化：

1. 增加 Reducer 内存（<font style="color:rgb(6, 10, 38);">mapreduce.reduce.memory.mb</font>）。
2. 调整内存分配策略。

数据倾斜（Data Skew）

表现：99% 的 Reducer 已完成，1 个 Reducer 还在跑（甚至 OOM）。

根源：某些 key（如 null、热门商品 ID）出现频率极高。

解决方案：

1. 预处理：过滤异常 key。
2. 加盐（Salting）：将热点 key 拆分为多个子 key，分散到不同 Reducer。
3. Hive 中可开启<font style="color:rgb(6, 10, 38);">hive.optimize.skewjoin=true</font>。

小文件问题（影响 Map 阶段）

大量小文件 → 启动过多 Mapper → 资源浪费 + 调度开销大。

优化：

1. 合并小文件（Hive:<font style="color:rgb(6, 10, 38);">set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;</font>）
2. 使用 SequenceFile / ORC / Parquet 等列式存储。