Hadoop MapReduce编程模型深度解析：从分片到输出的完整流程


🌺The Begin🌺点点关注，收藏不迷路🌺

引言

MapReduce是Hadoop的核心计算模型，它通过分而治之的思想，将大规模数据处理任务分解为可并行执行的Map和Reduce阶段。理解MapReduce的工作流程，是掌握Hadoop编程的基础。本文将基于你的描述，详细解析MapReduce的完整执行流程。

一、MapReduce整体流程概览

1.1 你的描述验证

你描述的流程非常准确，完整涵盖了MapReduce的各个阶段：

读取数据

键值对

分区数据

排序后

分组后

本地聚合

网络传输

最终结果

本地文件系统

Map阶段

Partition分区

Sort排序

Group分组

Combiner归约
可选

Shuffle传输

Reduce阶段

输出保存

二、Map阶段详解

2.1 输入分片与读取

public class MapStage {
    
    // 1. 输入分片（InputSplit）
    // MapReduce将输入文件切分成多个分片，每个分片由一个Map处理
    
    // 2. RecordReader读取数据
    // 将分片中的数据解析成键值对
    
    // 默认的TextInputFormat
    // 键: 行偏移量 (LongWritable)
    // 值: 行内容 (Text)
    
    // 输入文件内容示例
    /*
    hello world
    hadoop mapreduce
    */
    
    // 转换为键值对
    // <0, "hello world">
    // <12, "hadoop mapreduce">
}

2.2 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 line = value.toString();
        
        // 按空格分词
        String[] words = line.split(" ");
        
        // 输出每个单词，计数为1
        for (String w : words) {
            word.set(w);
            context.write(word, one);  // 输出 <word, 1>
        }
    }
}

2.3 Hadoop内置数据类型

Hadoop类型 Java类型 用途
LongWritable Long 键，通常表示行偏移量
IntWritable Integer 值，用于计数
Text String 文本数据
NullWritable null 占位符，无需实际值
DoubleWritable Double 浮点数

三、分区（Partition）阶段

3.1 默认HashPartitioner

public class HashPartitioner<K, V> extends Partitioner<K, V> {
    
    // 默认分区规则：根据key的hashCode取模
    public int getPartition(K key, V value, int numReduceTasks) {
        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
    
    // 例如：有3个Reducer，key="hello"
    // hashcode = 99162322
    // 99162322 % 3 = 1 → 分配到第1个Reducer
}

3.2 自定义分区器

public class CustomPartitioner extends Partitioner<Text, IntWritable> {
    
    @Override
    public int getPartition(Text key, IntWritable value, int numPartitions) {
        String word = key.toString();
        
        // 根据单词首字母分区
        char firstChar = word.charAt(0);
        
        if (firstChar >= 'a' && firstChar <= 'z') {
            return (firstChar - 'a') % numPartitions;
        } else {
            return 0;  // 其他字符统一分区
        }
    }
}

// 在Job中设置自定义分区器
job.setPartitionerClass(CustomPartitioner.class);

3.3 分区示意图

Map输出

hashCode%3=0

hashCode%3=1

hashCode%3=2

分区器

Reducer 1
apple, cat

Reducer 2
banana

Reducer 3
dog

四、排序（Sort）阶段

4.1 默认排序规则

// MapReduce默认按key进行字典序排序
// 输入: <apple,1>, <banana,1>, <cat,1>
// 排序后: <apple,1>, <banana,1>, <cat,1>

4.2 自定义排序

// 自定义数据类型实现WritableComparable
public class PersonWritable implements WritableComparable<PersonWritable> {
    
    private String name;
    private int age;
    
    @Override
    public int compareTo(PersonWritable o) {
        // 先按年龄排序，再按姓名排序
        if (this.age != o.age) {
            return this.age - o.age;  // 年龄升序
        } else {
            return this.name.compareTo(o.name);  // 姓名升序
        }
    }
}

// 或者继承WritableComparator
public class PersonComparator extends WritableComparator {
    
    protected PersonComparator() {
        super(PersonWritable.class, true);
    }
    
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        PersonWritable p1 = (PersonWritable) a;
        PersonWritable p2 = (PersonWritable) b;
        return p1.compareTo(p2);
    }
}

五、分组（Group）阶段

5.1 默认分组规则

// 默认情况下，分组与排序规则一致
// 相同key的值会被分到同一组

// 输入: <apple,1>, <apple,1>, <banana,1>
// 分组后:
// 组1: apple -> [1,1]
// 组2: banana -> [1]

5.2 自定义分组

// 自定义分组比较器
public class CustomGroupComparator extends WritableComparator {
    
    protected CustomGroupComparator() {
        super(Text.class, true);
    }
    
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        Text t1 = (Text) a;
        Text t2 = (Text) b;
        
        // 只按首字母分组
        String s1 = t1.toString();
        String s2 = t2.toString();
        
        char c1 = s1.charAt(0);
        char c2 = s2.charAt(0);
        
        return Character.compare(c1, c2);
    }
}

// 在Job中设置
job.setGroupingComparatorClass(CustomGroupComparator.class);

// 效果：apple和apricot会分到同一组，即使key不同

六、Combiner归约阶段

6.1 Combiner作用

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);
    }
}

// 在Job中设置
job.setCombinerClass(WordCountCombiner.class);

6.2 Combiner效果对比

阶段 无Combiner 有Combiner
Map输出 <apple,1>,<apple,1>,<apple,1> <apple,3>
Shuffle数据量 3条记录 1条记录
网络传输 3个单位 1个单位
Reduce计算量 需要累加3次 直接使用3

有Combiner

本地聚合

apple,3

1条记录

直接输出

Map

Combiner

Shuffle

Reduce

apple,3

无Combiner

apple,1
apple,1
apple,1

3条记录

累加

Map

Shuffle

Reduce

apple,3

七、Reduce阶段

7.1 Reducer实现

public class WordCountReducer 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);  // 输出最终结果
    }
}

7.2 Shuffle过程详解

Shuffle过程

分区

溢写

合并

拉取

合并排序

Map输出

内存缓冲区

本地磁盘
多个小文件

一个大文件
按分区排序

Reduce

Reducer输入

八、完整WordCount示例

8.1 主类实现

public class WordCount {
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        
        // 设置Jar包
        job.setJarByClass(WordCount.class);
        
        // 设置Mapper
        job.setMapperClass(WordCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        // 设置Combiner
        job.setCombinerClass(WordCountCombiner.class);
        
        // 设置Partitioner
        job.setPartitionerClass(HashPartitioner.class);
        
        // 设置Reducer
        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]));
        
        // 设置Reducer数量
        job.setNumReduceTasks(2);
        
        // 提交作业
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

九、执行流程总结

阶段 输入 处理 输出 关键组件
Map 输入分片 业务逻辑处理 中间键值对 Mapper
Partition 键值对 计算分区号 分区标记 Partitioner
Sort 分区数据 按key排序 排序后的数据 RawComparator
Group 排序后数据 合并相同key key-值列表 WritableComparator
Combiner 分组数据 本地聚合 聚合结果 Reducer
Shuffle Map输出 网络传输 Reduce输入 HTTP
Reduce 分组聚合 最终计算 最终结果 Reducer

十、总结

组件 作用 自定义方式
Mapper 数据处理 继承Mapper类
Partitioner 数据分区 继承Partitioner类
Comparator 排序规则 继承WritableComparator
GroupComparator 分组规则 继承RawComparator
Combiner 本地聚合 继承Reducer类
Reducer 最终聚合 继承Reducer类

核心要点：

  1. Map阶段：读取数据，转换为键值对，应用业务逻辑
  2. 分区：决定数据去哪个Reducer，默认HashPartitioner
  3. 排序：按key排序，可自定义排序规则
  4. 分组：相同key合并，可自定义分组规则
  5. Combiner：可选优化，减少网络传输
  6. Reduce：最终聚合计算，输出结果

一句话总结：MapReduce通过分而治之的思想，将大数据处理分解为Map（分）和Reduce（合）两个阶段，中间通过Shuffle连接，形成了一个完整的数据处理流水线。

在这里插入图片描述


🌺The End🌺点点关注，收藏不迷路🌺
# hadoop # mapreduce # 大数据
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

一个 API 入口调用多个大模型：AiiOnly客户端 + CC Switch 打通 Codex AI编程流程

avatar AI编程社区
cover

VSCode Claude Code + MiMo 中转免登录完整配置教程（解决原生登录弹窗 + 模型报错）

avatar AI编程社区
cover

从零开始做一个属于自己的 Skill（保姆级教程）

avatar AI编程社区