一、Map Task 与 Reduce Task的作用
1.1 Map Task和Reduce Task
1.1.1 Map Task（映射任务）
定义：处理输入数据分区的初始任务，执行
map、filter、flatMap等窄依赖转换操作。 【详细的宽窄依赖列举见第三点】特点：
- 每个Map Task独立处理一个输入分区
- 输出结果直接传递给下游任务或写入本地磁盘（Shuffle Write）
- 在Spark中对应
ShuffleMapTask类型
示例：
// Map Task处理逻辑valmapped=inputRDD.map(x=>(x.key,x.value))// 每个分区独立处理
1.1.2 Reduce Task（归约任务）
定义：从多个Map Task收集数据并进行聚合的任务，执行
reduceByKey、groupByKey、join等宽依赖操作。特点：
- 需要从多个Map Task拉取数据（Shuffle Read）
- 对相同key的数据进行聚合计算
- 在Spark中通常对应
ResultTask类型
示例：
// Reduce Task处理逻辑valreduced=mapped.reduceByKey(_+_)// 需要跨分区聚合
区别
| 特性 | Map Task | Reduce Task |
|---|---|---|
| 数据依赖 | 窄依赖（分区内处理） | 宽依赖（跨分区聚合） |
| 数据来源 | 输入数据或上游RDD | 多个Map Task的输出 |
| 网络传输 | 通常无（除非数据本地性不足） | 必须从多个节点拉取数据 |
| 输出目标 | 下游任务或本地磁盘 | 最终结果或下一阶段 |
| 任务类型 | ShuffleMapTask | ResultTask |
1.2 Spark的Stage1对比Stage2
| 特性 | Stage 0（Map阶段） | Stage 1（Reduce阶段） |
|---|---|---|
| 数据依赖 | 无，读取输入数据或上一个RDD | 依赖Stage 0的输出，需要shuffle read |
| 网络传输 | 通常无（除非数据本地性不足） | 需要从多个节点拉取数据 |
| 输出 | 写入本地磁盘（shuffle write） | 输出最终结果或传递给下一个操作 |
| 任务类型 | 通常是Map任务 | 通常是Reduce任务（但Spark中不严格区分） |
1.2.1 Stage 0（Map阶段）
// 示例：Stage 0的操作valrdd=sc.textFile("data.txt")// 读取数据.flatMap(_.split(" "))// 转换操作1.map(word=>(word,1))// 转换操作2// 这些操作都在Stage 0中，因为它们之间是窄依赖特点：
- 窄依赖：每个父RDD的分区最多被子RDD的一个分区使用
- 无需Shuffle：数据在同一个分区内处理
- 流水线优化：多个转换操作可以合并执行
1.2.2 Stage 1（Reduce阶段）
// 示例：Stage 1的操作valwordCounts=mappedRDD.reduceByKey(_+_)// 触发新的Stage二、Map Task 与 Reduce Task的案例分析
示例：统计海量文本中每个单词的出现次数
1. 输入数据
假设我们有三个文本片段（实际中可能是TB级数据）：
文本1: "hello world hello" 文本2: "hello mapreduce" 文本3: "world hello bigdata"2. MapReduce 执行流程
步骤1：分割（Split）
- 系统自动将输入文件分割成固定大小的分片（例如128MB），每个分片由一个Map 任务处理。
- 本例中，假设每个文本作为一个分片，共3个分片。
步骤2：Map 阶段（映射）
每个 Map 任务读取一个分片，并逐行处理。
Map 函数接收每一行文本，将其拆分为单词，并为每个单词输出中间键值对：
<单词, 1>。Map 函数伪代码：
defmap(line):forwordinline.split():emit(word,1)# 输出 (word, 1)Map 输出结果：
- Map1（处理文本1）：
(hello,1), (world,1), (hello,1) - Map2（处理文本2）：
(hello,1), (mapreduce,1) - Map3（处理文本3）：
(world,1), (hello,1), (bigdata,1)
- Map1（处理文本1）：
步骤3：Shuffle（洗牌）与排序
- 系统自动将所有 Map 输出中相同键的键值对分组，发送到同一个 Reduce 任务。
- 分组结果：
hello → [1, 1, 1, 1]（来自 Map1、Map2、Map3）world → [1, 1]mapreduce → [1]bigdata → [1]
步骤4：Reduce 阶段（归约）
每个 Reduce 任务接收一个键及其对应的值列表，进行聚合计算（如求和）。
Reduce 函数伪代码：
defreduce(word,counts):total=sum(counts)# 对计数列表求和emit(word,total)Reduce 输出结果：
- Reduce1 处理
hello：(hello, 4) - Reduce2 处理
world：(world, 2) - Reduce3 处理
mapreduce：(mapreduce, 1) - Reduce4 处理
bigdata：(bigdata, 1)
- Reduce1 处理
步骤5：输出结果
- 所有 Reduce 任务的输出合并为最终结果，写入分布式文件系统（如 HDFS）：
hello 4 world 2 mapreduce 1 bigdata 13. 图示流程
原始数据（分布到3个节点上）： 文本1 → "hello world hello" 文本2 → "hello mapreduce" 文本3 → "world hello bigdata" Map阶段（并行处理）： Map1 → (hello,1), (world,1), (hello,1) Map2 → (hello,1), (mapreduce,1) Map3 → (world,1), (hello,1), (bigdata,1) Shuffle阶段（自动分组）： hello → [1,1,1,1] world → [1,1] mapreduce → [1] bigdata → [1] Reduce阶段（并行聚合）： Reduce1 (hello) → 求和得4 → (hello,4) Reduce2 (world) → 求和得2 → (world,2) Reduce3 (mapreduce) → 求和得1 → (mapreduce,1) Reduce4 (bigdata) → 求和得1 → (bigdata,1) 最终输出： hello 4 world 2 mapreduce 1 bigdata 1三、附：DataFrame操作的依赖类型
3.1表格汇总
| 操作类型 | 具体操作 | 依赖类型 | 是否触发Shuffle | 说明 |
|---|---|---|---|---|
| 转换操作 | select() | 窄依赖 | 否 | 列投影，不改变数据分区 |
filter()/where() | 窄依赖 | 否 | 行过滤，不改变数据分区 | |
withColumn() | 窄依赖 | 否 | 添加/修改列，不改变分区 | |
withColumnRenamed() | 窄依赖 | 否 | 重命名列，不改变分区 | |
drop() | 窄依赖 | 否 | 删除列，不改变分区 | |
limit() | 窄依赖 | 否 | 限制行数，不改变分区 | |
union()/unionAll() | 窄依赖 | 否 | 合并DataFrame，分区数相加 | |
intersect() | 宽依赖 | 是 | 求交集，需要去重 | |
except() | 宽依赖 | 是 | 求差集，需要去重 | |
distinct() | 宽依赖 | 是 | 去重操作，需要全局比较 | |
dropDuplicates() | 宽依赖 | 是 | 删除重复行，需要Shuffle | |
| 聚合操作 | groupBy()+ 聚合函数 | 宽依赖 | 是 | 分组聚合必须Shuffle |
rollup() | 宽依赖 | 是 | 多维聚合，需要Shuffle | |
cube() | 宽依赖 | 是 | 全维度聚合，需要Shuffle | |
pivot() | 宽依赖 | 是 | 数据透视，需要Shuffle | |
| 排序操作 | orderBy()/sort() | 宽依赖 | 是 | 全局排序，需要Shuffle |
sortWithinPartitions() | 窄依赖 | 否 | 分区内排序，不Shuffle | |
| 重分区操作 | repartition() | 宽依赖 | 是 | 增加/重分布分区 |
coalesce() | 窄依赖 | 否 | 减少分区，不触发Shuffle | |
repartitionByRange() | 宽依赖 | 是 | 按范围重分区 | |
| Join操作 | join()(非广播) | 宽依赖 | 是 | 常规Join需要Shuffle |
join()(广播) | 窄依赖 | 否 | 广播小表，无Shuffle | |
crossJoin() | 宽依赖 | 是 | 笛卡尔积，需要Shuffle | |
| 窗口函数 | 窗口函数 +partitionBy() | 宽依赖 | 是 | 窗口函数通常需要Shuffle |
窗口函数 (无partitionBy) | 宽依赖 | 是 | 全局窗口需要Shuffle | |
| 其他操作 | sample() | 窄依赖 | 否 | 抽样，不改变分区 |
randomSplit() | 窄依赖 | 否 | 随机分割，不Shuffle | |
cache()/persist() | 窄依赖 | 否 | 缓存操作，不Shuffle | |
repartition(col) | 宽依赖 | 是 | 按列重分区 |
3.2 宽窄依赖分析例子
窄依赖操作示例
# 窄依赖操作链 - 无Shuffle，在一个Stage内执行df_transformed=(df.select("id","name","salary").filter(col("salary")>5000).withColumn("bonus",col("salary")*0.1).withColumnRenamed("id","employee_id"))宽依赖操作示例
# 宽依赖操作 - 触发Shuffle，创建新Stagedf_aggregated=(df.groupBy("department").agg(avg("salary").alias("avg_salary"),count("*").alias("employee_count")).orderBy("avg_salary"))混合依赖示例
# 混合依赖 - 会创建多个Stageresult=(df# Stage 1: 窄依赖操作.select("dept","name","salary","year").filter(col("year")==2023)# Stage 2: 宽依赖 (groupBy触发Shuffle).groupBy("dept").agg(avg("salary").alias("avg_salary"))# Stage 3: 宽依赖 (join触发Shuffle).join(dept_info,"dept","inner")# Stage 4: 窄依赖操作.select("dept","dept_name","avg_salary").orderBy("avg_salary"))四、性能优化建议
| 场景 | 优化策略 | 说明 |
|---|---|---|
| 减少Shuffle | 使用coalesce()代替repartition()减少分区 | coalesce避免全量Shuffle |
| 小表Join | 使用广播Join | 将小表广播到所有Executor |
| 多次使用中间结果 | 使用cache()/persist() | 避免重复计算 |
| 先过滤再处理原则 | 先过滤再处理原则，使用filter() | 减少后续处理的数据量 |
| 避免全局排序 | 用不需要shuffle的sortWithinPartitions()代替sort | 如果不需要全局有序 |
五、拓展：Shuffle Write底层原理
5.1 什么是Shuffle Write？
Shuffle Write是Map Task将处理后的数据按key分区并写入本地磁盘的过程。当Map Task完成计算后，需要为Reduce Task准备数据。
Shuffle Write的底层具体步骤：Hash分区、排序、内存溢写、合并文件
分区（Partitioning）
# 计算每个key对应的目标分区partition_id=hash(key)%num_reducers- 根据key的哈希值决定数据属于哪个Reduce分区
- 确保相同key的数据进入同一分区
排序（Sorting）
- 在每个分区内部按key排序
- Spark默认使用Sort-Shuffle机制
- 排序后相同key的数据连续存储，便于后续聚合
溢写（Spilling）
内存缓冲区 → 排序 → 写入磁盘临时文件- 当内存缓冲区满时，将排序好的数据写入磁盘
- 避免内存溢出导致任务失败
合并（Merging）
临时文件1 + 临时文件2 + ... → 合并文件（按分区+key排序）- 将多个临时文件合并为一个大文件
- 每个Map Task最终生成一个分区有序的输出文件
5.2 Shuffle传输与Shuffle Read
Shuffle传输过程
Map Task节点1 ──传输分区0数据──> Reduce Task节点A Map Task节点2 ──传输分区0数据──> Reduce Task节点A Map Task节点3 ──传输分区1数据──> Reduce Task节点B ...- 数据传输：每个Reduce Task从所有Map Task节点拉取属于自己的分区数据
- 网络协议：使用高效的二进制传输协议
- 流量控制：避免单个Reduce Task同时拉取过多数据导致网络拥塞
Shuffle Read（读取阶段）
Shuffle Read是Reduce Task从各个Map Task节点拉取并合并数据的过程：
数据拉取（Fetch）
- Reduce Task向Driver查询数据位置
- 并发从多个Map节点拉取数据块
- 支持本地读取优先（同一节点上的数据直接读取）
数据合并（Merge）
# 归并排序过程while还有未处理的数据块:从各数据流中取最小key的数据 相同key的数据合并 写入内存或输出- 将来自不同Map的同一分区数据合并排序
- 内存中维护合并缓冲区
聚合计算（Aggregation）
- 对于
reduceByKey等操作，在合并时直接执行聚合 - 对于
groupByKey，先分组再处理
- 对于
5.3 其他常见问题与解决方案
数据倾斜
- 现象：少数Reduce Task处理时间远长于其他
- 解决：使用
salting技术或broadcast join
网络瓶颈
- 现象：Shuffle Read时间过长
- 解决：调整
spark.reducer.maxSizeInFlight，启用数据压缩
内存不足
- 现象：频繁spill到磁盘
- 解决：增加Executor内存，调整
spark.shuffle.memoryFraction
