MapReduce案例

本文目录导读：

1. MapReduce 核心概念（3分钟速览）
2. 经典案例详解
3. 实战技巧与优化
4. Python 示例（Streaming模式）
5. 性能优化清单
6. 学习路径建议

下面我为你详细讲解 MapReduce 案例，从基础概念到典型案例分析，帮助你彻底理解这个分布式计算模型。

MapReduce 核心概念（3分钟速览）

MapReduce 由 Map（映射） 和 Reduce（归约） 两个核心阶段组成：

输入数据 → [Split] → [Map] → [Shuffle&Sort] → [Reduce] → 输出结果

经典案例详解

案例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

性能优化清单

学习路径建议

1. 手写WordCount → 理解基本流程
2. 尝试二次排序 → 深入Shuffle机制
3. 实现Join操作 → 学习复杂数据处理
4. 优化与调参 → 面试面试高分技巧
5. 对比Spark → 理解新一代计算框架优势

就是MapReduce的经典案例分析,从单词计数到搜索引擎索引，从简单统计到性能优化，这些案例覆盖了90%以上的实际业务场景。