大数据教程FG005-Hadoop MapReduce框架原理实战

本文档风哥主要介绍Hadoop MapReduce框架原理实战，包括MapReduce架构原理、MapReduce执行流程、MapReduce核心特性、MapReduce作业规划、MapReduce配置文件说明、MapReduce核心参数配置、MapReduce示例程序、MapReduce作业提交、MapReduce作业监控、WordCount单词统计案例、数据排序案例、MapReduce常见问题处理等内容，风哥教程参考Hadoop官方文档MapReduce Tutorial等内容，适合大数据开发人员在学习和测试中使用，如果要应用于生产环境则需要自行确认。

Part01-基础概念与理论知识

1.1 MapReduce架构原理

MapReduce是Hadoop的分布式计算框架，采用”分而治之”的思想，将大规模数据处理任务分解为Map和Reduce两个阶段。更多视频教程www.fgedu.net.cn

MapReduce架构特点：

Map阶段：数据分片、Map任务执行、中间结果输出
Shuffle阶段：数据分区、排序、分组、合并
Reduce阶段：数据汇总、Reduce任务执行、结果输出
容错机制：任务失败自动重试
数据本地性：计算向数据移动

1.2 MapReduce执行流程

MapReduce执行流程：

阶段	步骤	说明
输入	数据分片	InputFormat将输入数据切分为多个Split
Map	Map任务	每个Split对应一个Map任务，执行map()函数
Shuffle	分区排序	Map输出按Key分区、排序、分组
Reduce	Reduce任务	每个分区对应一个Reduce任务，执行reduce()函数
输出	结果输出	OutputFormat将结果写入HDFS

1.3 MapReduce核心特性

MapReduce核心特性：

自动并行化：自动将任务并行执行
容错性：任务失败自动重试
数据本地性：计算向数据移动，减少网络传输
扩展性：支持横向扩展
简单性：只需编写map()和reduce()函数

风哥提示：MapReduce是Hadoop的核心计算框架，理解MapReduce执行流程对于大数据开发至关重要。学习交流加群风哥微信: itpux-com

Part02-生产环境规划与建议

2.1 MapReduce作业规划

MapReduce作业规划需要考虑数据量、任务数量、资源分配等因素：

# MapReduce作业规划（示例）

## 数据量：100GB
## 文件数量：1000个文件
## 块大小：128MB

## Map任务数量计算
Map任务数 = 数据量 ÷ 块大小
Map任务数 = 100GB ÷ 128MB ≈ 800个

## Reduce任务数量计算
Reduce任务数 = 0.95 × NodeManager数量 × 最大容器数
Reduce任务数 = 0.95 × 6 × 3 ≈ 17个

## 资源配置
Map任务内存：2GB
Reduce任务内存：4GB
ApplicationMaster内存：2GB

## 总资源需求
Map资源 = 800 × 2GB = 1600GB
Reduce资源 = 17 × 4GB = 68GB
AM资源 = 2GB
总资源 = 1670GB

2.2 MapReduce配置文件说明

MapReduce核心配置文件：

# MapReduce配置文件说明

## mapred-site.xml
MapReduce核心配置文件，包括：
– mapreduce.framework.name：运行框架
– mapreduce.jobtracker.address：JobTracker地址
– mapreduce.job.maps：Map任务数量
– mapreduce.job.reduces：Reduce任务数量
– mapreduce.map.memory.mb：Map任务内存
– mapreduce.reduce.memory.mb：Reduce任务内存

## mapred-env.sh
MapReduce环境变量配置，包括：
– HADOOP_JOB_HISTORYSERVER_HEAPSIZE：历史服务器堆内存
– HADOOP_MAPRED_HOME：MapReduce安装目录

2.3 MapReduce核心参数配置

MapReduce核心参数配置建议：

# MapReduce核心参数配置

## mapred-site.xml配置
mapreduce.framework.name
yarn mapreduce.jobtracker.address
rm1:8032 mapreduce.job.maps
2 mapreduce.job.reduces
1 mapreduce.map.memory.mb
2048 mapreduce.reduce.memory.mb
4096 mapreduce.map.java.opts
-Xmx1638m mapreduce.reduce.java.opts
-Xmx3276m mapreduce.task.io.sort.mb
256 mapreduce.task.io.sort.factor
100 mapreduce.reduce.shuffle.parallelcopies
50

生产环境建议：MapReduce内存配置需要根据数据量和任务复杂度进行调整，建议Map任务内存设置为2GB-4GB，Reduce任务内存设置为4GB-8GB。学习交流加群风哥QQ113257174

Part03-生产环境项目实施方案

3.1 MapReduce示例程序

3.1.1 运行示例程序

# 准备测试数据
echo “hello world hello hadoop” > /tmp/input.txt
echo “hello mapreduce hello yarn” >> /tmp/input.txt
echo “hadoop mapreduce spark” >> /tmp/input.txt

# 上传测试数据到HDFS
hdfs dfs -mkdir -p /user/hadoop/input
hdfs dfs -put /tmp/input.txt /user/hadoop/input/

# 运行WordCount示例程序
hadoop jar $HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.5.jar wordcount /user/hadoop/input /user/hadoop/output

2026-04-08 15:00:30,123 INFO client.RMProxy: Connecting to ResourceManager at rm1/192.168.1.15:8032
2026-04-08 15:00:30,456 INFO input.FileInputFormat: Total input files to process : 1
2026-04-08 15:00:30,789 INFO mapreduce.JobSubmitter: number of splits:1
2026-04-08 15:00:31,012 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1234567890123_0001
2026-04-08 15:00:31,234 INFO impl.YarnClientImpl: Submitted application application_1234567890123_0001
2026-04-08 15:00:31,456 INFO mapreduce.Job: The url to track the job: http://rm1:8088/proxy/application_1234567890123_0001/
2026-04-08 15:00:31,678 INFO mapreduce.Job: Running job: job_1234567890123_0001
2026-04-08 15:00:40,890 INFO mapreduce.Job: Job job_1234567890123_0001 running in uber mode : false
2026-04-08 15:00:40,012 INFO mapreduce.Job: map 0% reduce 0%
2026-04-08 15:00:50,234 INFO mapreduce.Job: map 100% reduce 0%
2026-04-08 15:01:00,456 INFO mapreduce.Job: map 100% reduce 100%
Job Ended: 1234567890

# 查看输出结果
hdfs dfs -cat /user/hadoop/output/part-r-00000

hadoop 2
hello 4
mapreduce 2
spark 1
world 1
yarn 1

3.2 MapReduce作业提交

3.2.1 提交作业到指定队列

# 提交作业到prod队列
hadoop jar $HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.5.jar wordcount -D mapreduce.job.queuename=prod /user/hadoop/input /user/hadoop/output2

# 提交作业并指定Map和Reduce任务数量
hadoop jar $HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.5.jar wordcount -D mapreduce.job.maps=10 -D mapreduce.job.reduces=5 /user/hadoop/input /user/hadoop/output3

# 提交作业并指定内存配置
hadoop jar $HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.5.jar wordcount -D mapreduce.map.memory.mb=4096 -D mapreduce.reduce.memory.mb=8192 /user/hadoop/input /user/hadoop/output4

3.3 MapReduce作业监控

3.3.1 查看作业状态

# 查看作业列表
mapred job -list

2026-04-08 15:10:30,123 INFO client.RMProxy: Connecting to ResourceManager at rm1/192.168.1.15:8032
Total jobs:1
JobId State StartTime UserName Queue Priority UsedContainers RsvdContainers UsedMem RsvdMem NeededMem TotalMaps TotalReduces MapsCompleted ReducesCompleted
job_1234567890123_0001 RUNNING 1234567890 hadoop default NORMAL 2 0 6144MB 0MB 6144MB 1 1 1 0

# 查看作业详情
mapred job -status job_1234567890123_0001

2026-04-08 15:10:35,123 INFO client.RMProxy: Connecting to ResourceManager at rm1/192.168.1.15:8032
Job file: hdfs://nn1:9000/user/hadoop/.staging/job_1234567890123_0001/job.xml
Job tracker: ‘rm1:8032’
Start time: 1234567890123
End time: 0
Average map time: 10s
Average reduce time: 5s
Counter: 23
Job Counters
Launched map tasks=1
Launched reduce tasks=1
Data-local map tasks=1
Total time spent by all maps in occupied slots (ms)=10000
Total time spent by all reduces in occupied slots (ms)=5000
Total time spent by all map tasks (ms)=10000
Total time spent by all reduce tasks (ms)=5000
Total vcore-milliseconds taken by all map tasks=10000
Total vcore-milliseconds taken by all reduce tasks=5000
Total megabyte-milliseconds taken by all map tasks=10240000
Total megabyte-milliseconds taken by all reduce tasks=5120000
Map-Reduce Framework
Map input records=3
Map output records=9
Map output bytes=120
Map output materialized bytes=130
Input split bytes=100
Combine input records=9
Combine output records=6
Reduce input groups=6
Reduce shuffle bytes=130
Reduce input records=6
Reduce output records=6
Spilled Records=12
Shuffled Maps =1
Failed Shuffles=0
Merged Map outputs=1
GC time elapsed (ms)=100
CPU time spent (ms)=1000
Physical memory (bytes) snapshot=268435456
Virtual memory (bytes) snapshot=2147483648
Total committed heap usage (bytes)=134217728
File Input Format Counters
Bytes Read=60
File Output Format Counters
Bytes Written=50

风哥提示：MapReduce作业监控是生产环境运维的重要环节，需要关注作业执行时间、资源使用情况、失败任务等信息。更多学习教程公众号风哥教程itpux_com

Part04-生产案例与实战讲解

4.1 WordCount单词统计案例

4.1.1 WordCount原理

# WordCount执行流程

## 输入数据
hello world hello hadoop
hello mapreduce hello yarn
hadoop mapreduce spark

## Map阶段
输入：(行号, 行内容)
输出：(单词, 1)

(hello, 1)
(world, 1)
(hello, 1)
(hadoop, 1)
(hello, 1)
(mapreduce, 1)
(hello, 1)
(yarn, 1)
(hadoop, 1)
(mapreduce, 1)
(spark, 1)

## Shuffle阶段
分区：按单词首字母分区
排序：按单词排序
分组：相同单词分为一组

(hadoop, [1, 1])
(hello, [1, 1, 1, 1])
(mapreduce, [1, 1])
(spark, [1])
(world, [1])
(yarn, [1])

## Reduce阶段
输入：(单词, [1, 1, …])
输出：(单词, 总数)

(hadoop, 2)
(hello, 4)
(mapreduce, 2)
(spark, 1)
(world, 1)
(yarn, 1)

4.1.2 WordCount代码示例

# WordCount Java代码示例

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

public static class TokenizerMapper extends Mapper {

private final static IntWritable one = new IntWritable(1);
private Text word = new Text();

public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
String[] tokens = value.toString().split(“\\s+”);
for (String token : tokens) {
word.set(token);
context.write(word, one);
}
}
}

public static class IntSumReducer extends Reducer {

private IntWritable result = new IntWritable();

public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
context.write(key, result);
}
}

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(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.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);
}
}

4.2 数据排序案例

4.2.1 数据排序原理

# 数据排序执行流程

## 输入数据
5
3
8
1
9
2

## Map阶段
输入：(行号, 数字)
输出：(数字, 空值)

(5, )
(3, )
(8, )
(1, )
(9, )
(2, )

## Shuffle阶段
排序：按数字排序

(1, )
(2, )
(3, )
(5, )
(8, )
(9, )

## Reduce阶段
输入：(数字, [空值, …])
输出：(数字, 空值)

1
2
3
5
8
9

4.3 MapReduce常见问题处理

4.3.1 任务失败重试

# 问题现象：MapReduce任务失败重试
# 分析步骤：

# 1. 查看任务日志
yarn logs -applicationId application_1234567890123_0001

# 2. 查看任务失败原因
mapred job -history /user/hadoop/output/_logs/history/job_1234567890123_0001.jhist

# 3. 检查资源配置
mapred job -status job_1234567890123_0001

# 解决方案：
# – 增加任务内存
# – 检查数据倾斜
# – 检查代码逻辑
# – 检查数据格式

4.3.2 数据倾斜问题

# 问题现象：Reduce任务执行缓慢
# 分析步骤：

# 1. 查看Reduce任务进度
mapred job -status job_1234567890123_0001

# 2. 查看数据分布
hdfs dfs -cat /user/hadoop/input/* | awk ‘{print $1}’ | sort | uniq -c | sort -nr

# 解决方案：
# – 使用Combiner减少数据传输
# – 增加Reduce任务数量
# – 使用自定义分区器
# – 使用二次排序

Part05-风哥经验总结与分享

5.1 MapReduce最佳实践

MapReduce最佳实践：

合理设置任务数量：根据数据量和集群资源设置Map和Reduce任务数量
使用Combiner：减少Map端数据传输
压缩中间结果：减少Shuffle数据量
避免小文件：合并小文件，减少Map任务数量
监控作业执行：及时发现和解决问题

5.2 MapReduce性能优化建议

MapReduce性能优化建议：

# MapReduce性能优化建议

## Map阶段优化
– 增加Map任务内存
– 使用Combiner减少数据传输
– 压缩Map输出
– 调整IO缓冲区大小

## Shuffle阶段优化
– 增加Shuffle并行度
– 调整排序缓冲区大小
– 使用压缩减少网络传输

## Reduce阶段优化
– 增加Reduce任务内存
– 调整Reduce任务数量
– 使用二次排序优化数据倾斜

## 资源配置优化
– 合理配置任务内存
– 调整JVM参数
– 优化GC策略

5.3 MapReduce作业检查清单

MapReduce作业检查清单：

MapReduce作业检查清单：

□ 输入数据已上传到HDFS
□ 输出目录不存在
□ 资源配置合理
□ 队列配置正确
□ 作业提交成功
□ 作业执行正常
□ 输出结果正确
□ 监控告警已配置

风哥提示：MapReduce是Hadoop的核心计算框架，合理的作业规划和性能优化对于大数据处理至关重要。from bigdata视频:www.itpux.com

本文由风哥教程整理发布,仅用于学习测试使用,转载注明出处:http://www.fgedu.net.cn/10327.html