MapReduce里Reducer如何把千万条数据合并成100条——从词频统计到数据去重原理结构与常见bug全解析
你见过凌晨三点还在跑MapReduce作业的大数据工程师吗？
我见过。就在昨天，一个做日志分析的团队，他们的作业从晚上八点开始跑，到了凌晨三点还在shuffle阶段挣扎。Reducer一直在把几百个Mapper的输出合并，内存报警、 spill阈值触发、归并排序反复进行……最后终于把几千亿条日志数据压缩成了不到两百条聚合结果。
这听起来像神话，但MapReduce就是这么干的。
今天我们就来拆解这件事：Reducer是怎么把千万条数据合并成几百条、甚至一百条的？ 我们从一个最经典的词频统计说起，一路走到数据去重，把原理、结构、代码、踩过的坑全部讲透。
一、先建立直觉：Reducer是什么角色
想象你在主持一场大型合唱团比赛。
有100个合唱队参赛，每个队50个人。你的任务是什么？不是记住每个人的名字，而是——统计每个队有多少人、每支队伍的平均音高是多少、哪支队伍唱得最大声。
MapReduce里的Reducer就是这样的角色。Mapper是各个合唱队的指挥，他们已经把本队的数据预处理好了（比如每人唱了什么音符，计数是多少）。Reducer要做的是：接收所有Mapper发来的结果，按照规则合并、排序、输出最终答案。
千万条数据变成100条，本质上就是一场巨大的”按组聚合”操作。
二、词频统计：最经典的教学案例
2.1 问题描述
给定一个巨大的文本文件（比如1TB的网页日志），统计每个单词出现的次数，最终输出出现频率最高的100个单词及其频次。
2.2 完整MapReduce流程
输入：1TB的文本文件
输出：Top 100高频词及其词频
2.3 Mapper阶段的逻辑
每个Mapper负责读取文本的一个split（比如64MB或128MB），逐行扫描，遇到单词就输出：
<单词, 1>
比如一行文本是：
hello world hello mapreduce hello world
Mapper会输出：
hello 1
world 1
hello 1
mapreduce 1
hello 1
world 1
关键点：Mapper不合并，它只拆分和初步计数。 每个单词出现一次就输出一个<key, 1>。
2.4 Shuffle阶段：数据的”大迁徙”
这是整个MapReduce最核心的环节。Mapper的输出不会直接到Reducer，而是经过Shuffle：
Mapper输出 --> 本地排序 --> 溢写(Spill) --> Merge合并 --> Reducer接收
Shuffle的具体过程：
Step 1：环形缓冲区
Mapper的输出先写入一个内存中的环形缓冲区（默认100MB）。当缓冲区使用率达到80%时，触发溢写：
// Java伪代码：环形缓冲区管理
private byte[] buffer = new byte[1024 * 1024 * 100]; // 100MB
private int bufferStartPtr; // 缓冲区起始位置
private int bufferEndPtr; // 缓冲区已用位置
private int bufferLowWaterMark; // 低水位线(80%)
private int bufferHighWaterMark; // 高水位线
void write(String key, int value) {
// 如果缓冲区满了，触发spill
if (bufferEndPtr >= bufferHighWaterMark) {
spill();
}
// 写入缓冲区
writeStringToBuffer(key);
writeIntToBuffer(value);
}
Step 2：溢写到本地磁盘
当缓冲区达到阈值，数据被溢写到本地磁盘，形成一个个小的排序文件：
void spill() {
// 按key排序（快排）
Arrays.sort(records, comparator);
// 写入磁盘
FileSystem fs = FileSystem.get(conf);
Path spillPath = new Path("/tmp/spill_" + System.currentTimeMillis());
FSDataOutputStream out = fs.create(spillPath);
for (Record r : sortedRecords) {
out.write(r.key.getBytes());
out.write(r.value);
}
out.close();
}
Step 3：Merge合并
如果有多个溢写文件（比如跑了5次spill），Reducer端需要对它们进行多路归并排序：
void mergeMultipleSpills(List<Path> spillFiles) {
// 为每个spill文件创建输入迭代器
List<InputIterator> iterators = new ArrayList<>();
for (Path p : spillFiles) {
iterators.add(new InputIterator(p));
}
// 最小堆归并排序
PriorityQueue<InputIterator> minHeap = new PriorityQueue<>();
for (InputIterator it : iterators) {
if (it.hasNext()) {
minHeap.add(it);
}
}
// 输出合并结果
while (!minHeap.isEmpty()) {
InputIterator min = minHeap.poll();
emit(min.next());
if (min.hasNext()) {
minHeap.add(min);
}
}
}
2.5 Reducer阶段的逻辑
Reducer收到所有Mapper按key排序后的数据，进行聚合：
public class WordCountReducer extends Reducer<Text, IntWritable, Text, 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();
}
context.write(key, new IntWritable(sum));
}
}
一个关键细节：Reducer是按key分组处理的。 所有相同的key会被分组到一起，Reducer对每个组执行一次reduce函数。
三、千万条变100条：数据压缩的本质
3.1 为什么数据量会急剧减少？
回到词频统计的例子。假设1TB的文本中：
- 总共有10亿个单词（千万条
<单词, 1>记录） - 不同的单词只有5000个
Mapper输出：10亿条记录 Reducer输入：10亿条记录（经过shuffle） Reducer输出：5000条记录（每个单词一行）
这就是”千万条变100条”的本质：大量重复的key被合并成了少量的key。
3.2 数据分布规律
现实中的数据往往符合长尾分布（Zipf定律）：
排名1的单词"the"：出现1000万次
排名2的单词"be"：出现800万次
排名3的单词"and"：出现600万次
...
排名100的单词：出现约10万次
排名10000的单词：出现约100次
排名1000000的单词：出现约1次
所以：
- 高频词：只有几百个，但占据了大部分数据量
- 低频词：数量庞大，但每个词只出现几次
Reducer需要处理的是：把同一个key的所有value聚合起来，输出一个（key, aggregated_value）对。
四、从词频统计到数据去重：两种不同的Reducer策略
4.1 词频统计：Reducer做聚合
输入到Reducer的每条记录：
<wikipedia, 1>
<wikipedia, 1>
<wikipedia, 1>
<大数据, 1>
<大数据, 1>
...
Reducer处理：
- 对所有<wikipedia, 1>求和 → <wikipedia, 1000000>
- 对所有<大数据, 1>求和 → <大数据, 500000>
- ...
输出：
<wikipedia, 1000000>
<大数据, 500000>
...
4.2 数据去重：Reducer做”标记”
数据去重的问题描述：从海量数据中去掉重复项，保留唯一值。
原始数据（1000万条）：
user_id
1001
1002
1001
1003
1002
1001
...
期望输出（去重后）：
1001
1002
1003
...
方法一：Mapper中去重，Reducer透传
public class DeduplicateMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
private Text outKey = new Text();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 每个Mapper内部用HashSet去重
String line = value.toString();
context.write(new Text(line), NullWritable.get());
}
}
public class DeduplicateReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
@Override
protected void reduce(Text key, Iterable<NullWritable> values, Context context)
throws IOException, InterruptedException {
// 因为key已经排序，相同key相邻
// 直接输出第一个即可
context.write(key, NullWritable.get());
}
}
方法二：只让Reducer做去重（更简洁）
public class DeduplicateMapper extends Mapper<LongWritable, Text, Text, Text> {
private Text outKey = new Text();
private Text outValue = new Text("1");
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 输出 <去重字段, 标记>
context.write(value, outValue);
}
}
public class DeduplicateReducer extends Reducer<Text, Text, Text, NullWritable> {
@Override
protected void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
// 不管values有多少个，只输出一次key
// 这就是"千万条变100条"的关键
context.write(key, NullWritable.get());
}
}
原理剖析：
Mapper输出：
<user_1001, 1>
<user_1002, 1>
<user_1001, 1> ← 重复
<user_1003, 1>
<user_1002, 1> ← 重复
...
经过Shuffle后，Reducer按key排序并分组：
组1: <user_1001, [1, 1, 1, 1, ...]> ← 可能有一万条
组2: <user_1002, [1, 1, 1, ...]> ← 可能有五千条
组3: <user_1003, [1, 1, ...]> ← 可能有三百条
...
Reducer对每组只输出一个key：
user_1001
user_1002
user_1003
...
这就是Reducer实现数据去重的核心：它不关心value是多少，只关心key是否重复。相同的key在Shuffle阶段已经被合并到一起，Reducer只需要对每组输出一个代表即可。
五、Reducer的工作原理深度解析
5.1 Reduce任务的生命周期
1. Copy阶段（Shuffle）
├── 从每个Mapper复制输出到本地
├── 内存中的缓冲区管理
└── 当缓冲区满时溢写到磁盘
2. Sort阶段
├── 对内存中的数据排序
├── 多路归并排序合并溢写文件
└── 按key分组
3. Reduce阶段
├── 调用用户的reduce函数
├── 输出结果
└── 清理资源
5.2 内存管理：Reducer如何容纳千万条数据？
这是最容易被忽视的问题。Reducer的内存是有限的：
// 关键配置参数
mapreduce.reduce.memory.mb=4096 // Reducer可用内存4GB
mapreduce.reduce.java.opts=-Xmx3072m // JVM堆内存3GB
mapreduce.task.io.sort.mb=1024 // 环形缓冲区1GB
mapreduce.task.io.sort.factor=100 // 同时merge的文件数
mapreduce.reduce.shuffle.parallelcopies=5 // 并行copy的Mapper数量
内存模型：
┌─────────────────────────────────────────┐
│ Reducer JVM Heap (3GB) │
├─────────────────────────────────────────┤
│ Shuffle Buffer (1GB) │
│ ├── 内存中待排序的数据 │
│ └── 达到阈值后溢写到磁盘 │
├─────────────────────────────────────────┤
│ Sort Spill (磁盘) │
│ ├── 多个溢写文件 │
│ └── 归并排序后交给Reducer处理 │
├─────────────────────────────────────────┤
│ Output Buffer (128MB) │
│ └── 输出结果写入HDFS │
└─────────────────────────────────────────┘
关键点：Reducer不需要把所有数据都放在内存里！ 它采用”流式处理”模式：
- Mapper的数据不断拷贝到Reducer
- 内存满了就溢写到磁盘
- 归并排序后，按key分组交给用户reduce函数
- reduce函数处理完一组数据后，释放内存，继续处理下一组
// 这就是为什么Reducer可以处理比内存大得多的数据
// 因为它是"边收边处理"的流式架构
5.3 分组与排序
Reducer接收到的数据是按key排序的，这意味着相同的key会相邻出现：
输入到Reducer（已排序）：
<a, 1>
<a, 3>
<a, 2>
<b, 5>
<b, 1>
<c, 7>
...
Reducer的分组处理：
组1: key=a, values=[1, 3, 2] → reduce(a, [1,3,2]) → 输出 a:6
组2: key=b, values=[5, 1] → reduce(b, [5,1]) → 输出 b:6
组3: key=c, values=[7] → reduce(c, [7]) → 输出 c:7
分组的关键配置：
// 自定义分组器（默认按key的自然排序分组）
job.setGroupingComparatorClass(MyGroupingComparator.class);
// 自定义比较器（控制排序和分组的逻辑）
job.setSortComparatorClass(MySortComparator.class);
六、从词频统计到Top 100：实际压缩策略
6.1 问题升级：只需要输出Top 100高频词
之前的词频统计会输出所有单词的词频。但如果我们只需要Top 100，Reducer可以进一步优化：
public class Top100Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {
// 维护一个大小为100的最小堆
private PriorityQueue<Entry> minHeap = new PriorityQueue<>(100);
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
// 计算该key的总词频
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
// 如果堆未满，直接加入
if (minHeap.size() < 100) {
minHeap.add(new Entry(key.toString(), sum));
} else {
// 如果当前词频大于堆顶（最小值），替换堆顶
Entry minEntry = minHeap.peek();
if (sum > minEntry.getFrequency()) {
minHeap.poll();
minHeap.add(new Entry(key.toString(), sum));
}
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
// 最后输出堆中的所有元素（需要额外排序）
List<Entry> result = new ArrayList<>(minHeap);
result.sort((a, b) -> b.getFrequency() - a.getFrequency());
for (Entry entry : result) {
context.write(new Text(entry.getWord()),
new IntWritable(entry.getFrequency()));
}
}
}
这就是”千万条变100条”的实际例子：
- Mapper输出：10亿条
<单词, 1> - Shuffle后：仍然是10亿条，但按单词排序
- Reducer处理：每个单词聚合为
<单词, 总频次>，约5000条 - Top100优化：最终只输出100条
6.2 Combiner：Reducer的”前置压缩”
Combiner是一个本地Reducer，在Mapper端运行，提前合并相同key的数据，大幅减少Shuffle的数据量：
// Combiner和Reducer逻辑相同
public class WordCountCombiner 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);
}
}
使用Combiner后的数据流：
Mapper输出（无Combiner）：
<wikipedia, 1>
<wikipedia, 1>
<wikipedia, 1>
<wikipedia, 1>
...
→ 100万条记录发送到Shuffle
Mapper输出（有Combiner）：
<wikipedia, 10000>
<wikipedia, 5000>
<wikipedia, 8000>
...
→ 只有几百条记录发送到Shuffle！
Combiner的规则：
- 只能用于交换律和结合律的操作（求和、求最大值、求最小值）
- 不能用于求平均值、中位数等需要全局信息的操作
- Combiner的输出格式必须和Reducer的输入格式一致
七、数据去重的进阶：多种策略对比
7.1 策略一：Mapper本地去重 + Reducer透传
public class DedupMapper1 extends Mapper<LongWritable, Text, Text, NullWritable> {
private Set<String> seen = new HashSet<>();
private Text outKey = new Text();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
if (!seen.contains(line)) {
seen.add(line);
outKey.set(line);
context.write(outKey, NullWritable.get());
}
}
}
优缺点：
- 优点：大幅减少Shuffle数据量
- 缺点：每个Mapper的HashSet会占用内存，大数据量下内存压力大
7.2 策略二：Mapper不复用 + Reducer去重（推荐）
public class DedupMapper2 extends Mapper<LongWritable, Text, Text, Text> {
private Text outKey = new Text();
private Text outValue = new Text("1");
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
outKey.set(value);
context.write(outKey, outValue);
}
}
public class DedupReducer2 extends Reducer<Text, Text, Text, NullWritable> {
@Override
protected void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
// 不管values有多少，只输出一次key
context.write(key, NullWritable.get());
}
}
优缺点：
- 优点：逻辑简单，内存占用小
- 缺点：Shuffle数据量较大（但Reducer处理效率高）
7.3 策略三：二次排序去重（精确去重）
当数据分布极不均匀时（比如某些key出现亿次，某些key只出现几次），可以使用二次排序：
// 自定义Key，包含原始数据和辅助排序字段
public class DedupKey implements WritableComparable<DedupKey> {
private Text data; // 原始数据
private IntWritable seq; // 辅助排序字段
@Override
public void write(DataOutput out) throws IOException {
data.write(out);
seq.write(out);
}
@Override
public void readFields(DataInput in) throws IOException {
data.readFields(in);
seq.readFields(in);
}
@Override
public int compareTo(DedupKey o) {
int cmp = data.compareTo(o.data);
if (cmp != 0) return cmp;
return seq.compareTo(o.seq);
}
}
八、常见Bug与陷阱
8.1 Bug 1：Reducer内存溢出（OOM）
现象： Reducer任务失败，日志显示Java heap space或GC overhead limit exceeded
原因：
1. 某个key的数据量远超预期（数据倾斜）
2. Reduce函数中存储了过多数据
3. 自定义的Value对象占用了大量内存
解决方案：
// 1. 调整Reducer内存配置
job.setMapReduceJobName("myjob");
conf.setInt("mapreduce.reduce.memory.mb", 8192);
conf.setInt("mapreduce.reduce.java.opts", "-Xmx6144m");
// 2. 在Reduce函数中避免大对象缓存
// 错误做法：把全部values存入List
List<IntWritable> allValues = new ArrayList<>();
for (IntWritable val : values) {
allValues.add(val); // 可能导致OOM
}
// 正确做法：流式聚合
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
context.write(key, new IntWritable(sum));
// 3. 使用Combiner减少Shuffle数据量
job.setCombinerClass(WordCountCombiner.class);
8.2 Bug 2：数据丢失（Key被截断）
现象： 输出结果比预期少，某些key的数据丢失了
原因：
// 错误：重复使用同一个Text对象
public class BadReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
private Text resultKey = new Text(); // 静态复用！
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
resultKey.set(key); // 问题：所有记录指向同一个对象！
context.write(resultKey, new IntWritable(sum));
}
}
解决方案：
// 正确：每次创建新的对象，或使用key的副本
public class GoodReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
// 每次创建新的Text对象
context.write(new Text(key), new IntWritable(sum));
}
}
8.3 Bug 3：Reducer输出顺序不符合预期
现象： 输出的数据没有按key排序
原因：
// MapReduce保证的是：相同key的记录被分组成一个reduce调用
// 但不保证Reducer之间的输出顺序
// 如果需要全局有序，需要使用TotalOrderPartitioner
解决方案：
// 使用TotalOrderPartitioner确保全局有序
job.setPartitionerClass(TotalOrderPartitioner.class);
// 生成分割点文件
TotalOrderPartitioner.setPartitionFile(conf, new Path("/user/output/partitions"));
// 或者确保只有一个Reducer（小数据集可用）
job.setNumReduceTasks(1);
8.4 Bug 4：Combiner使用不当导致结果错误
现象： 使用了Combiner后，输出结果不正确
原因：
// 错误：求平均值时使用Combiner
public class BadAverageCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
private IntWritable result = new IntWritable();
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
int count = 0;
for (IntWritable val : values) {
sum += val.get();
count++;
}
result.set(sum / count); // 错误！局部平均值不能直接传给Reducer
context.write(key, result);
}
}
解决方案：
// 正确：Combiner输出(sum, count)对，Reducer计算最终平均值
public class GoodAverageCombiner extends Reducer<Text, IntWritable, Text, Text> {
private Text result = new Text();
protected void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int sum = 0;
int count = 0;
for (IntWritable val : values) {
sum += val.get();
count++;
}
result.set(sum + "," + count);
context.write(key, result);
}
}
// Reducer计算全局平均值
public class AverageReducer extends Reducer<Text, Text, Text, DoubleWritable> {
private DoubleWritable result = new DoubleWritable();
protected void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
int totalSum = 0;
int totalCount = 0;
for (Text val : values) {
String[] parts = val.toString().split(",");
totalSum += Integer.parseInt(parts[0]);
totalCount += Integer.parseInt(parts[1]);
}
result.set((double) totalSum / totalCount);
context.write(key, result);
}
}
8.5 Bug 5：Reducer输出文件数量过多
现象： 输出了几百个文件，每个文件只有几行数据
原因： Reducer数量太多，或者数据分布极不均匀
解决方案：
// 1. 调整Reducer数量
job.setNumReduceTasks(10); // 根据数据量和输出文件大小调整
// 2. 使用自定义Partitioner控制数据分布
job.setPartitionerClass(MyPartitioner.class);
// 3. 合并小文件（在作业完成后）
public class MergeSmallFiles {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
Path input = new Path(args[0]);
Path output = new Path(args[1]);
// 读取所有小文件，合并后重新写入
SequenceFile.Writer writer = null;
try {
writer = new SequenceFile.Writer(fs, conf, output,
Text.class, Text.class);
FileStatus[] files = fs.listStatus(input);
for (FileStatus file : files) {
if (file.isFile()) {
SequenceFile.Reader reader = new SequenceFile.Reader(fs, conf, file.getPath());
Text key = new Text();
Text value = new Text();
while (reader.next(key, value)) {
writer.append(key, value);
}
reader.close();
}
}
} finally {
if (writer != null) writer.close();
}
}
}
8.6 Bug 6：数据倾斜导致单个Reducer处理过长时间
现象： 大部分Reducer很快完成，但某个Reducer跑了几个小时
原因： 某个key的数据量远超其他key（比如一个热门词出现了1亿次）
解决方案：
// 1. 使用二次Key散列
public class SkewDataMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
private Text outKey = new Text();
private IntWritable outValue = new IntWritable(1);
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String word = value.toString();
// 判断是否为热点key（词频超过阈值）
if (isHotWord(word)) {
// 热点key加上随机后缀，分散到多个Reducer
int suffix = new Random().nextInt(10);
outKey.set(word + "_" + suffix);
} else {
outKey.set(word);
}
context.write(outKey, outValue);
}
}
// 2. 在Reducer中处理散列key
public class SkewDataReducer extends Reducer<Text, IntWritable, Text, 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();
}
// 去掉散列后缀
String cleanKey = key.toString().split("_")[0];
context.write(new Text(cleanKey), new IntWritable(sum));
}
}
九、性能优化技巧
9.1 调整Map和Reduce任务数量
// 计算最优的Mapper数量
long fileSize = fs.getContentSummary(inputPath).getLength();
long blockSize = fs.getDefaultBlockSize(inputPath);
int mapperNum = (int) (fileSize / blockSize) + 1;
job.setNumMapTasks(mapperNum);
// 计算最优的Reducer数量
long outputSizePerReducer = 100 * 1024 * 1024; // 每个Reducer输出100MB
long totalOutputSize = estimateOutputSize();
int reducerNum = (int) (totalOutputSize / outputSizePerReducer) + 1;
job.setNumReduceTasks(Math.max(reducerNum, 1));
9.2 启用压缩减少Shuffle数据量
// 启用Map输出压缩
job.setBoolean("mapreduce.map.output.compress", true);
job.setClass("mapreduce.map.output.compress.codec",
SnappyCodec.class, CompressionCodec.class);
// 启用Reducer输出压缩
job.setBoolean("mapreduce.output.fileoutputformat.compress", true);
job.setClass("mapreduce.output.fileoutputformat.compress.codec",
SnappyCodec.class, CompressionCodec.class);
job.set("mapreduce.output.fileoutputformat.compress.type",
RecordOutputStreamer.BLOCK);
9.3 调整Shuffle参数
// 增大环形缓冲区
conf.setInt("mapreduce.task.io.sort.mb", 512); // 默认100MB，调大到512MB
// 降低溢写阈值（更早溢写，减少内存压力）
conf.setFloat("mapreduce.task.io.sort.factor", 100);
// 增大Reducer拷贝并行度
conf.setInt("mapreduce.reduce.shuffle.parallelcopies", 10);
十、总结：千万条变100条的核心原理
让我用一个表格来总结整个过程：
| 阶段 | 数据量变化 | 关键操作 |
|---|---|---|
| 原始输入 | 1TB文本 | 分片（Split） |
| Mapper输出 | 10亿条<词,1> |
扫描、输出 |
| Combiner处理后 | 1亿条<词,频次> |
本地聚合 |
| Shuffle后 | 1亿条（按词排序） | 网络传输、排序 |
| Reducer聚合后 | 5000条<词,总频次> |
分组聚合 |
| Top100筛选后 | 100条 | 堆排序、筛选 |
核心要点：
- Reducer不是”一次性”处理所有数据的，而是流式处理，边收边排边输出
- Shuffle阶段的数据排序和分组是Reducer能够压缩数据的关键
- Combiner可以大幅减少Shuffle的数据量，但只能用于满足结合律的操作
- 数据倾斜是Reducer性能问题的头号杀手，需要特殊处理
- 内存管理是MapReduce最精妙的设计，通过溢写和归并排序，让Reducer能够处理比内存大得多的数据
十一、给初学者的建议
如果你是刚接触MapReduce，建议按以下步骤学习：
1. 先手写一个最简单的WordCount程序（本地模式运行）
2. 理解Mapper、Shuffle、Reducer的完整数据流
3. 添加Combiner，观察数据量的变化
4. 尝试数据去重场景
5. 处理TopN问题（使用堆数据结构）
6. 研究数据倾斜的解决方案
7. 学习Hadoop的调优参数
记住：MapReduce的核心思想不是”计算”，而是”如何将大规模数据的问题分解成可并行处理的小问题，再安全地合并结果”。Reducer只是这个过程中的最后一个合并步骤。千万条数据变成100条，本质上就是把”重复的key合并成唯一的key，把分散的value聚合成聚合的value”。
理解了这一点，你就理解了MapReduce的精髓。
这篇文章基于Hadoop 2.x / 3.x的MapReduce模型编写，核心原理适用于所有MapReduce实现（包括Spark的Reduce操作、Flink的GroupReduce等）。