> **摘要**：Reduce Task 如何找到成百上千个 Map Task 输出的 Shuffle 文件？如何精准定位到目标分区的数据块？本文从 MapOutputTracker 全局注册表、MapStatus 元数据结构、IndexShuffleBlockResolver 的 Index→Data 解析链、Netty 远程 Block 拉取、ExternalShuffleService 退出守护五个维度，配合 2 张架构图 + 源码追踪，完整拆解 Spark Shuffle 文件寻址的端到端流程。
>
> **关键词**：MapOutputTracker, MapStatus, ShuffleBlockId, IndexShuffleBlockResolver, BlockManager, ExternalShuffleService

---

## 一、开篇：Reduce Task 的寻址困境

假设一个 ShuffleDependency 有 **1000 个 Map Task**，200 个 Reduce 分区。Reduce Task-5 需要从 **1000 个不同的 Executor** 上拉取属于自己分区的数据。它是怎么知道去哪个节点、读取哪个文件、文件中的哪一段数据的？

```
Shuffle 寻址三要素
① 哪个节点？→ MapOutputTracker 查 MapStatus.location
② 哪个文件？→ shuffle_X_Y_0.data
③ 哪段数据？→ IndexFile[offset+length] → DataFile.seek(offset)
```

---

## 二、MapOutputTracker 架构全景

![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-2482016/8a66dee1b30371614fcbcf9983259507.png)

### 2.1 MapOutputTrackerMaster — 全局注册表

```scala
class MapOutputTrackerMaster(conf: SparkConf) {
  // key: shuffleId, value: Array[MapStatus] 按 mapId 索引
  val mapStatuses = new ConcurrentHashMap[Int, Array[MapStatus]]()

  // Map Task 完成时调用
  def registerMapOutput(shuffleId: Int, mapId: Int, status: MapStatus): Unit

  // Reduce Task 查询时调用
  def getMapSizesByExecutorId(shuffleId: Int, reduceId: Int)
      : Seq[(BlockManagerId, Seq[(BlockId, Long)])]
}
```

### 2.2 MapStatus — 每个 Map Task 的寻址元数据

```scala
case class MapStatus(location: BlockManagerId, partitionSizes: Array[Long])
// location         → Executor 的 host + port + executorId
// partitionSizes   → 每个 Reduce 分区的压缩后大小
//                   (Reduce Task 通过前 N-1 个分区大小累加得到第 N 个分区 offset)
```

### 2.3 MapOutputTrackerWorker — Executor 端本地缓存

```scala
// Reduce Task 调用入口
BlockStoreShuffleReader.read():
  val statuses = mapOutputTracker.getMapSizesByExecutorId(shuffleId, reduceId)
  // → 返回 (BlockManagerId, Seq[(ShuffleBlockId, dataSize)]) 列表
  // → BlockManager.getBlockData() 逐一拉取
```

---

## 三、Index File → Data File 解析链
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-2482016/b22f101c9c5c2cc9441e8b81ac1ee2ac.png)

### 3.1 三步走：BlockId → Offset → Data

```scala
// ① 构造 BlockId
val blockId = ShuffleBlockId(shuffleId, mapId, reduceId)

// ② IndexShuffleBlockResolver.getBlockData(blockId)
val indexFile = getIndexFile(shuffleId, mapId)   // shuffle_X_Y_0.index
indexFile.seek(reduceId * 16L)                     // 定位到分区索引
val offset = indexFile.readLong()                  // 8 字节 offset
val length = indexFile.readLong()                  // 8 字节 length

// ③ 读取 data file
val dataFile = getDataFile(shuffleId, mapId)       // shuffle_X_Y_0.data
dataFile.seek(offset); dataFile.readFully(buf, 0, length)
```

### 3.2 IndexShuffleBlockResolver 源码

```scala
override def getBlockData(blockId: ShuffleBlockId): ManagedBuffer = {
  val indexFile = getIndexFile(blockId.shuffleId, blockId.mapId)
  val in = new DataInputStream(new FileInputStream(indexFile))
  in.skip(blockId.reduceId * 16L)            // 跳过前 reduceId 个分区索引
  val offset = in.readLong()                  // ← 关键
  val nextOffset = in.readLong()              // 下一个分区的 offset
  in.close()
  new FileSegmentManagedBuffer(getDataFile(...), offset, nextOffset - offset)
}
// reduceId × 16 定位 → 一次 seek + read → 返回 ManagedBuffer
```

---

## 四、远程拉取：Netty + BlockManager

```
Reduce 端 ShuffleClient.fetchBlocks()
  → NettyBlockTransferService.fetchBlocks(host, port, execId, blockIds)
    → Netty 连接远端 BlockManager
    → 服务端 NettyBlockRpcServer.receive(OpenBlocks)
      → BlockManager.getBlockData(blockId)
        → IndexShuffleBlockResolver.getBlockData()
    → FileRegion.transferTo() 发送数据块（零拷贝）

重试: spark.shuffle.io.maxRetries=3
退避: spark.shuffle.io.retryWait=5s
```

---

## 五、ExternalShuffleService：Executor 退出后的守护

```
正常模式（无 ESS）:
  Executor 退出 → Shuffle 数据丢失 → 重算 Map Task

ExternalShuffleService（ESS）:
  Executor 退出 → ESS 仍持有 shuffle 文件
  → MapOutputTracker 中的 MapStatus.location 指向 ESS
  → Reduce Task 直接从 ESS 拉取

配置: spark.shuffle.service.enabled=true
端口: spark.shuffle.service.port=7337
```

---

## 六、寻址公式总结

```
Reduce Task 端到端寻址流程:
① getMapSizesByExecutorId(shuffleId, reduceId)
   → MapOutputTrackerWorker 向 Driver 请求 MapStatus 列表
② 遍历每个 MapStatus:
   Node = status.location
   BlockId = ShuffleBlockId(shuffleId, mapId, reduceId)
   Size = status.partitionSizes(reduceId)
③ fetchBlocks(Node, Seq(BlockId))
   → Netty → 远端 BlockManager → getBlockData(BlockId)
④ getBlockData = IndexFile.seek(reduceId × 16) → offset+length → DataFile
⑤ 解压 → 反序列化 → ExternalAppendOnlyMap → Iterator
```

---

## 七、总结

1. **MapOutputTracker**：Driver 端全局注册表 + Executor 端本地缓存。Map Task 完成后注册 MapStatus → Reduce Task 获取 (节点, BlockId, 大小) 列表。

2. **Index File 精确定位**：每分区 16B 索引，`reduceId × 16` 一次 seek 即获 offset+length，无需全量扫描。

3. **ExternalShuffleService**：YARN 下 Executor 退出后数据不丢失，减少 Map Task 重算。


摘要：Reduce Task 如何找到成百上千个 Map Task 输出的 Shuffle 文件？如何精准定位到目标分区的数据块？本文从 MapOutputTracker 全局注册表、MapStatus 元数据结构、IndexShuffleBlockResolver 的 Index→Data 解析链、Netty 远程 Block 拉取、ExternalShuffleService 退出守护五个维度，配合 2 张架构图 + 源码追踪，完整拆解 Spark Shuffle 文件寻址的端到端流程。

Spark 核心之 Shuffle 文件寻址详解

摘要：Reduce Task 如何找到成百上千个 Map Task 输出的 Shuffle 文件？如何精准定位到目标分区的数据块？本文从 MapOutputTracker 全局注册表、MapStatus 元数据结构、IndexShuffleBlockResolver 的 Index→Data 解析链、Netty 远程 Block 拉取、ExternalShuffleService 退出守护五个维度

编程语言

MapOutputTracker全局注册表驱动Shuffle文件寻址，MapStatus记录节点与分区大小，IndexShuffleBlockResolver通过IndexFile的16字节索引精确定位DataFile偏移量，Netty远程拉取配合ExternalShuffleService保障Executor退出后数据可用。

数据丢失

数据结构

代码解释

Spark 

Yarn

让每一个 Agent，都跑在云上

bigdata-class

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

MapOutputTracker全局注册表驱动Shuffle文件寻址，MapStatus记录节点与分区大小，IndexShuffleBlockResolver通过IndexFile的16字节索引精确定位DataFile偏移量，Netty远程拉取配合ExternalShuffleService保障Executor退出后数...

Spark 核心之 Shuffle 文件寻址详解-腾讯云开发者社区-腾讯云

Spark 核心之 Shuffle 文件寻址详解

Spark 核心之 Shuffle 文件寻址详解

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐