本文目录导读：
下面我为你详细讲解 MapReduce 案例，从基础概念到典型案例分析，帮助你彻底理解这个分布式计算模型。
MapReduce 核心概念（3分钟速览）
MapReduce 由 Map（映射） 和 Reduce（归约） 两个核心阶段组成：
输入数据 → [Split] → [Map] → [Shuffle&Sort] → [Reduce] → 输出结果| 阶段 | 输入 | 输出 | 处理逻辑 |
|---|---|---|---|
| Map | <K1,V1> |
<K2,V2> |
提取信息，打标签 |
| Reduce | <K2,List<V2>> |
<K3,V3> |
聚合计算 |
经典案例详解
案例1：单词计数（WordCount）—— "Hello World"级入门
场景：统计大规模文本文件中每个单词出现的次数。
Java 代码实现
// Mapper 类
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[] words = value.toString().split("\\s+");
for (String w : words) {
word.set(w);
context.write(word, one); // 输出 <单词, 1>
}
}
}
// Reducer 类
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));
}
}
// 主驱动类
public class WordCount {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(WordCountMapper.class);
job.setCombinerClass(WordCountReducer.class); // 可选，本地合并优化
job.setReducerClass(WordCountReducer.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);
}
}
执行流程演示
输入: "Hello World Hello Hadoop"
→ Split: ["Hello", "World", "Hello", "Hadoop"]
→ Map输出: <Hello,1> <World,1> <Hello,1> <Hadoop,1>
→ Shuffle&Sort: <Hadoop,[1]> <Hello,[1,1]> <World,[1]>
→ Reduce: <Hadoop,1> <Hello,2> <World,1>案例2： 逆向索引（Inverted Index）—— 搜索引擎基础
场景：统计每个单词出现在哪些文档中，构建“单词→文档列表”的索引。
核心代码（Mapper和Reducer）
// Mapper
public class IndexMapper extends Mapper<LongWritable, Text, Text, Text> {
private Text word = new Text();
private Text docId = new Text();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 假设输入格式: "文档ID: 内容"
String[] parts = value.toString().split(":", 2);
String doc = parts[0].trim();
String content = parts[1].trim();
String[] words = content.split("\\s+");
for (String w : words) {
word.set(w);
docId.set(doc);
context.write(word, docId); // 输出 <单词, 文档ID>
}
}
}
// Reducer
public class IndexReducer extends Reducer<Text, Text, Text, Text> {
@Override
protected void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
Set<String> docs = new HashSet<>();
for (Text val : values) {
docs.add(val.toString());
}
context.write(key, new Text(docs.toString()));
}
}
数据示例
输入:
doc1: apple banana apple
doc2: banana orange
doc3: apple orange
输出:
apple [doc1, doc3]
banana [doc1, doc2]
orange [doc2, doc3]案例3：最高温度统计 —— 数值聚合经典
场景：从气象数据中找出每年的最高温度。
关键处理逻辑
// Mapper: 提取年份和温度
public class MaxTempMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> {
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
// 假设格式: "年份 温度"
String[] parts = line.split("\\s+");
if (parts.length == 2) {
context.write(new Text(parts[0]), new DoubleWritable(Double.parseDouble(parts[1])));
}
}
}
// Reducer: 计算最大值
public class MaxTempReducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
@Override
protected void reduce(Text key, Iterable<DoubleWritable> values, Context context)
throws IOException, InterruptedException {
double max = Double.MIN_VALUE;
for (DoubleWritable val : values) {
max = Math.max(max, val.get());
}
context.write(key, new DoubleWritable(max));
}
}
案例4：Top-N 问题 —— 部分聚合优化
场景：从海量数据中找出得分最高的前10条记录。
特点
- Map阶段局部Top-N：每个Map只输出该分区的Top-N
- Reduce阶段全局Top-N：汇总所有局部Top-N，再排序
// Mapper: 每个Map输出当前分片的Top-N
public class TopNMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
private TreeMap<Long, String> topN = new TreeMap<>();
private static final int N = 10;
@Override
protected void map(LongWritable key, Text value, Context context) {
String[] parts = value.toString().split("\t");
if (parts.length == 2) {
long score = Long.parseLong(parts[1]);
topN.put(score, parts[0]);
if (topN.size() > N) {
topN.remove(topN.firstKey()); // 移除最小的
}
}
}
@Override
protected void cleanup(Context context) {
// 只输出该Map的Top-N
for (Map.Entry<Long, String> entry : topN.entrySet()) {
try {
context.write(new Text(entry.getValue()), new LongWritable(entry.getKey()));
} catch (Exception e) { }
}
}
}
实战技巧与优化
Combiner 优化（本地聚合）
在Map端先做一次合并,减少Shuffle数据量：
job.setCombinerClass(WordCountReducer.class); // 复用Reducer逻辑
数据倾斜处理
- 加盐：给热键添加随机前缀拆分
- 二次聚合：两轮MapReduce
自定义Partitioner
public class MyPartitioner extends Partitioner<Text, IntWritable> {
@Override
public int getPartition(Text key, IntWritable value, int numReduceTasks) {
// 根据key哈希分区
return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}
}
// 配置
job.setPartitionerClass(MyPartitioner.class);
job.setNumReduceTasks(4);
Python 示例（Streaming模式）
# mapper.py
#!/usr/bin/env python
import sys
for line in sys.stdin:
words = line.strip().split()
for word in words:
print(f"{word}\t1")
# reducer.py
#!/usr/bin/env python
import sys
from collections import Counter
counter = Counter()
for line in sys.stdin:
word, count = line.strip().split('\t')
counter[word] += int(count)
for word, count in counter.items():
print(f"{word}\t{count}")
# 运行命令
hadoop jar hadoop-streaming.jar \
-input input_dir \
-output output_dir \
-mapper mapper.py \
-reducer reducer.py \
-file mapper.py -file reducer.py
性能优化清单
| 优化点 | 操作 | 收益 |
|---|---|---|
| 压缩 | 开启输出压缩 (snappy/gzip) | 减少IO 50%+ |
| Combiner | 本地聚合 | 减少Shuffle网络流量 |
| 并行度 | 合理设置Map/Reduce数量 | 提高集群利用率 |
| 内存 | 调整JVM堆内存 (mapreduce.map.memory.mb) | 防止OOM |
| 推测执行 | 开启推测任务 | 避免慢节点拖累 |
学习路径建议
- 手写WordCount → 理解基本流程
- 尝试二次排序 → 深入Shuffle机制
- 实现Join操作 → 学习复杂数据处理
- 优化与调参 → 面试面试高分技巧
- 对比Spark → 理解新一代计算框架优势
就是MapReduce的经典案例分析,从单词计数到搜索引擎索引，从简单统计到性能优化，这些案例覆盖了90%以上的实际业务场景。
