第6章 MapReduce基础与编程
MapReduce 怎么把一个大任务拆成很多小任务、Shuffle 到底做了什么、两个案例代码怎么落地。
本章学习目标
- 能说清 MapReduce 的分布式并行计算思想与特点
- 能完整描述 MapReduce 的工作流程（切片→Map→Shuffle→Reduce）
- 能独立完成单词统计程序的设计、编码与运行
- 能完成二次排序案例并说明其设计思路
本章导览
核心内容
MapReduce概述
MapReduce 的核心思想是"分而治之 + 计算向数据靠拢"：把大任务拆成互不依赖的小任务并行执行，再把结果汇总；同时尽量把计算放到数据所在的节点上，减少网络传输。
分布式并行计算
- 数据被切成多个分片（InputSplit），每个分片交给一个 Map 任务
- Map 任务在数据所在节点本地执行，避免搬运数据
- 多个 Reduce 并行处理分区后的中间结果
- 框架负责调度、容错与重试，用户只写 map() 与 reduce()
MapReduce的特点
| 优点 | 局限 |
|---|---|
| 易于编程：只写两个函数 | 不适合实时计算（秒级延迟做不到） |
| 良好扩展性：加机器即扩容 | 不适合流式计算（数据要先落盘） |
| 高容错：任务失败自动重跑 | 不适合有向图/迭代计算（多轮落盘慢） |
| 适合海量数据离线批处理 | 不适合大量小文件 |
MapReduce的工作流程
一次作业的完整链路：<b>输入分片 → Map → Shuffle（分区/排序/合并）→ Reduce → 输出</b>。其中 Shuffle 是性能关键，也是调优的主要战场。
输入分片和数据格式化
- InputSplit 是逻辑分片，默认与 block 大小一致（128MB）
- InputFormat 决定"怎么切、怎么读"，常用 TextInputFormat 按行读
- RecordReader 把分片解析成 <key, value> 交给 Mapper
mapreduce.input.fileinputformat.split.maxsize。Shuffle过程
Shuffle 横跨 Map 端与 Reduce 端，是 MapReduce 中数据流动最复杂的阶段。
| 阶段 | 做了什么 |
|---|---|
| Map 端：溢写 | 结果先写环形缓冲区（默认 100MB，80% 触发溢写），溢写到磁盘 |
| Map 端：分区排序 | 按 key 分区（Partitioner）、分区内按 key 排序 |
| Map 端：合并 | 多个溢写文件合并（Combiner 可提前聚合，减少数据量） |
| Reduce 端：拉取 | 各 Reduce 从所有 Map 拉取自己分区的数据 |
| Reduce 端：归并 | 拉取的数据做归并排序，形成有序的 <key, 迭代器> |
💡 调优三板斧：① 加 Combiner 减少传输；② 调大缓冲区/溢写阈值；③ 合理设置 Reduce 数（默认 1，公式见 6.3.6）。
Reduce过程
- 对每个 key 调用一次 reduce()，处理该 key 的所有 value
- 输出通过 OutputFormat 写入 HDFS（默认 part-r-00000 等文件）
- Reduce 数决定输出文件数：几个 Reduce 就有几个 part 文件
案例解析——单词统计编程
教材把这个案例拆成"案例描述 → 设计思路 → 执行过程 → 编程实现 → 运行程序 → 数据优化"六步，是 MapReduce 的标准讲法。
编程实现（Mapper/Reducer/Driver）
// Mapper：每行拆词，输出 (word, 1)
public class WcMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
private Text k = new Text();
private IntWritable v = new IntWritable(1);
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] words = value.toString().split("\\s+");
for (String w : words) {
if (w.isEmpty()) continue;
k.set(w);
context.write(k, v); // (hello,1)
}
}
}
// Reducer：同一 key 的 value 累加
public class WcReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
private IntWritable out = new IntWritable();
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable v : values) sum += v.get();
out.set(sum);
context.write(key, out); // (hello,2)
}
}
// Driver：组装作业
public class WcDriver {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://hadoop1:9000");
Job job = Job.getInstance(conf, "wordcount");
job.setJarByClass(WcDriver.class);
job.setMapperClass(WcMapper.class);
job.setReducerClass(WcReducer.class);
job.setCombinerClass(WcReducer.class); // 关键：本地先聚合
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
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);
}
}运行程序
# 1) 打成 jar 并上传
mvn clean package
scp target/wc.jar hadoop@hadoop1:~/
# 2) 准备数据并提交
hdfs dfs -mkdir -p /wc/input
hdfs dfs -put w.txt /wc/input/
hadoop jar wc.jar com.demo.WcDriver /wc/input /wc/output
# 3) 看结果
hdfs dfs -cat /wc/output/part-r-00000数据优化
- 加 Combiner 减少网络传输
- 调 Reduce 个数：<code class="inl">reduce数 ≈ 输入数据量 / 每个Reduce处理量（默认 256MB）</code>
- 开启 Map 端压缩：<code class="inl">mapreduce.map.output.compress=true</code>
- 小文件先合并，避免过多 Map 任务
- 数据倾斜时给 key 加随机前缀做两阶段聚合
案例解析——二次排序编程
二次排序 = 先按第一列排序，第一列相同时再按第二列排序。核心是把两个字段组合成一个可比较的 key。
编程实现（自定义 WritableComparable + 分区 + 分组）
// 1) 组合 key：年+温度，实现比较规则
public class YearTempKey implements WritableComparable<YearTempKey> {
private int year; private int temp;
@Override public int compareTo(YearTempKey o) {
int r = Integer.compare(year, o.year); // 先按年
return r != 0 ? r : Integer.compare(o.temp, this.temp); // 同年按温度降序
}
// write/readFields 省略
}
// 2) 分区器：按年分区，保证同一年进入同一 Reduce
public class YearPartitioner extends Partitioner<YearTempKey, Text> {
@Override public int getPartition(YearTempKey k, Text v, int num) {
return Math.abs(k.getYear()) % num;
}
}
// 3) 分组比较器：同一个年进入同一次 reduce 调用
job.setGroupingComparatorClass(YearGroupingComparator.class);
job.setPartitionerClass(YearPartitioner.class);💡 二次排序的三个配件：组合 key（排序规则） + Partitioner（进哪个 Reduce） + GroupingComparator（谁归一组）。这一套也是 MapReduce 实现 TopN 的基础。
本章小结
本章从"分而治之 + 计算向数据靠拢"的思想出发，讲清 MapReduce 的工作流程（切片 → Map → Shuffle → Reduce），重点剖析 Shuffle 的溢写、分区、排序、合并与拉取归并，随后用单词统计和二次排序两个案例完成"设计思路 → 编程实现 → 运行 → 优化"的完整闭环。
思考与练习 先自己做，再点开答案
Map 任务的数量由什么决定？
查看参考答案
由输入分片（InputSplit）数量决定。分片默认与 HDFS block 大小一致（128MB），可通过 split.maxsize 等参数调整，一个分片对应一个 Map 任务。
为什么 Combiner 不能用于求平均值？
查看参考答案
Combiner 要求运算满足结合律与交换律。平均数不满足——先各自求平均再平均，结果与整体平均不同；它只能在 Map 端做"局部聚合"，不能改变整体语义。
Reduce 个数设置得越多越好吗？
查看参考答案
不是。Reduce 太少会导致单任务数据量大、耗时长；太多则产生大量小文件、增加调度开销。经验公式：Reduce 数 ≈ 输入数据量 ÷ 每个 Reduce 的建议处理量（默认 256MB），一般控制在集群可用槽位的 0.95 倍以内。
二次排序需要哪三个关键组件？
查看参考答案
① 自定义 WritableComparable 组合 key（定义排序规则）；② Partitioner（决定同组数据进哪个 Reduce）；③ GroupingComparator（决定哪些 key 归入同一次 reduce 调用）。