上一篇解决了 Tungsten 的内存管理和代码生成优化。这一篇进入存储体系——Spark 如何管理分布在集群中的数据块。

BlockManager 容易被忽略，因为用户代码很少直接和它打交道。但 RDD 的 persist/cache、Shuffle 的中间文件、广播变量的分发、累加器的聚合，底层全部通过 BlockManager 完成。它是 Spark 存储层的统一入口。

本文只抓一个问题：BlockManager 如何在内存和磁盘之间管理数据块，以及广播变量和累加器在这个体系中的位置。

BlockManager 架构

每个 Executor 上运行一个 BlockManager 实例，Driver 上也有一个（主要用于接收广播变量和累加器结果）。BlockManager 内部包含三个核心组件：

1
2
3
4
BlockManager
├── MemoryStore     内存存储（堆内/堆外）
├── DiskStore       磁盘存储（本地目录）
└── BlockTransferService  网络传输（Netty）

MemoryStore 负责在 JVM 堆内或堆外内存中存储数据块。DiskStore 负责把数据块序列化后写入本地磁盘目录（spark.local.dir 配置的路径）。BlockTransferService 负责从远程 Executor 拉取数据块。

每个数据块有一个全局唯一的 BlockId，格式取决于数据类型：

1
2
3
rdd_<rddId>_<partitionIndex>      RDD 分区缓存
shuffle_<shuffleId>_<mapId>_<reduceId>   Shuffle 数据
broadcast_<broadcastId>_<piece>   广播变量分片

Driver 上的 BlockManagerMaster 维护所有 BlockId 到位置（哪个 Executor）的映射。当一个 Task 需要读取远程 Block 时，先向 BlockManagerMaster 查询位置，再从目标 Executor 的 BlockTransferService 拉取。

RDD 缓存机制

调用 rdd.persist(storageLevel) 或 rdd.cache()（等价于 persist(MEMORY_ONLY)）时，Spark 并不立即缓存数据。它只是在 RDD 上打一个标记。当 action 触发计算时，每个 Task 计算完分区数据后，检查该 RDD 是否有 persist 标记。如果有，把计算结果交给 BlockManager 存储。

StorageLevel 控制存储行为：

1
2
3
4
5
6
7
MEMORY_ONLY          只存内存，放不下就不存
MEMORY_AND_DISK      先存内存，内存不够溢写到磁盘
MEMORY_ONLY_SER      序列化后存内存（更紧凑，但读取需要反序列化）
MEMORY_AND_DISK_SER  序列化后存内存+磁盘
DISK_ONLY            只存磁盘
*_2                  以上任一级别 + 副本（存两份在不同节点）
OFF_HEAP             堆外内存（不受 GC 影响）

MEMORY_ONLY 和 MEMORY_AND_DISK 的差别在于淘汰策略：MEMORY_ONLY 在内存不足时直接丢弃最早缓存的分区（下次用到时通过 Lineage 重算），MEMORY_AND_DISK 在内存不足时把最早的分区溢写到磁盘（下次读取从磁盘加载，比重算快）。

实际选择的经验法则：如果数据量能放进内存，用 MEMORY_ONLY。如果放不进但重算代价大（长 Lineage 或外部数据源），用 MEMORY_AND_DISK。如果 GC 压力大，用 MEMORY_ONLY_SER（序列化后一个分区就是一个字节数组，比散装 Java 对象对 GC 友好得多）。

广播变量：Torrent 分发

广播变量（Broadcast Variable）用于把一份只读数据高效地分发到所有 Executor。典型场景是 Map-Side Join——把小表广播到所有节点，避免 Shuffle。

1
2
3
4
val lookupTable = Map("a" -> 1, "b" -> 2, "c" -> 3)
val bc = sc.broadcast(lookupTable)

rdd.map(key => (key, bc.value.getOrElse(key, 0)))

如果不用广播变量，lookupTable 会随每个 Task 序列化发送一次。如果有 1000 个 Task，同一份数据就传输 1000 次。广播变量只传输一次到每个 Executor，然后所有 Task 共享同一份内存中的副本。

Spark 使用类似 BitTorrent 的 Torrent 分发协议来广播数据：

1
2
3
4
5
6
7
8
9
10
11
12
13
传统方式（Driver → 每个 Executor）:
  Driver ──→ Executor 0
  Driver ──→ Executor 1
  Driver ──→ Executor 2
  Driver ──→ ...
  瓶颈: Driver 的网络带宽

Torrent 方式:
  Driver ──→ Executor 0
  Driver ──→ Executor 1    Executor 0 ──→ Executor 3
  Driver ──→ Executor 2    Executor 1 ──→ Executor 4
                            Executor 2 ──→ Executor 5
  已收到数据的 Executor 也成为数据源

广播数据被切成若干个 4MB 的分片（spark.broadcast.blockSize）。每个 Executor 接收到一个分片后，立即成为该分片的数据源，其他 Executor 可以从它那里拉取。这样 Driver 的网络带宽不再是瓶颈，整个集群的聚合带宽被充分利用。

累加器：Driver 端聚合

累加器（Accumulator）是从 Executor 向 Driver 单向聚合数据的机制。典型用途是计数器和求和器。

1
2
3
4
5
6
7
8
9
val errorCount = sc.longAccumulator("errors")

rdd.foreach { record =>
  if (record.contains("ERROR")) {
    errorCount.add(1)
  }
}

println(s"Error count: ${errorCount.value}")

累加器的语义有一个关键陷阱：在 transformation 中使用累加器时，如果 Task 重试或 Stage 重算，累加器的值可能被重复累加。

1
2
3
4
5
6
安全：在 action 内部使用（如 foreach、reduce）
  └→ 每个 Task 只执行一次，结果准确

不安全：在 transformation 内部使用（如 map、filter）
  └→ Task 重试时同一条记录可能被累加多次
  └→ Stage 重算时整个分区的累加值重复

Spark 只保证在 action 中累加器值的准确性。在 transformation 中使用累加器的结果是"尽力而为"的近似值，不能作为精确计数。

实验：观察存储行为

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
// 缓存一个 RDD 并观察 Storage 页面
val rdd = sc.parallelize(1 to 1000000, 10).map(i => (i, i * i))
rdd.persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK)
rdd.count()  // 触发缓存

// Spark UI → Storage 页面
// 可以看到 RDD 的缓存大小、分区数、内存/磁盘占用

// 广播变量
val bigMap = (1 to 100000).map(i => (i, s"value_$i")).toMap
val bc = sc.broadcast(bigMap)
println(s"Broadcast size: ~${bigMap.size * 20 / 1024}KB")

// 累加器
val counter = sc.longAccumulator("myCounter")
rdd.foreach { case (k, v) => if (k % 2 == 0) counter.add(1) }
println(s"Even count: ${counter.value}")

在 Spark UI 的 Storage 页面可以看到缓存的 RDD，包括每个分区在哪个 Executor 上、占用了多少内存和磁盘空间。Environment 页面可以查看 spark.storage.memoryFraction 等存储相关配置。

模式提炼

BlockManager 体现了统一存储抽象模式：把不同类型的数据（缓存、Shuffle、广播）统一到同一个存储接口下，用 BlockId 作为寻址键，用 StorageLevel 控制存储策略。

这个模式在分布式存储系统中很常见。HDFS 的 NameNode 维护文件到 Block 的映射和 Block 到 DataNode 的映射，角色类似 BlockManagerMaster。对象存储（S3/OSS）用 key 寻址对象，不区分数据类型。

广播变量的 Torrent 分发体现了另一个模式：P2P 数据分发。已经拥有数据的节点成为新的数据源，避免单点瓶颈。CDN 的回源策略和 Docker 镜像的 P2P 分发（如 Dragonfly）使用同样的思路。

工程迁移表

Spark 概念 MapReduce 对应 Flink 对应 数据库对应
BlockManager 无统一存储层 TaskManager 的 MemoryManager Buffer Pool Manager
MemoryStore / DiskStore Map 输出写本地磁盘 Network Buffer + Spill 内存页 + 磁盘页
BlockManagerMaster 无（Job 间不共享存储元数据） 无直接对应 元数据服务（Catalog）
广播变量 DistributedCache Broadcast State 物化视图分发
累加器 Counter（MapReduce Counter） Accumulator（Flink Accumulator） 聚合函数的分布式部分聚合
StorageLevel 无（总是磁盘） 无（由 State Backend 决定） 缓冲区淘汰策略（LRU/Clock）

常见误解

误解一：“cache() 之后数据一定在内存中”。cache() 等价于 persist(MEMORY_ONLY)。如果 Executor 内存不足以存储所有分区，多余的分区不会被缓存——下次用到时通过 Lineage 重算。只有 MEMORY_AND_DISK 才保证所有分区都被持久化（放不进内存的写磁盘）。

误解二：“广播变量可以修改”。广播变量在 Executor 端是只读的。如果需要在 Executor 上修改共享状态并回传给 Driver，应该用累加器。但累加器只支持加法语义（或自定义的 AccumulatorV2 的 merge 语义），不支持任意修改。

误解三：“unpersist() 会立即释放内存”。unpersist(blocking=false)（默认）是异步的，不保证立即释放。如果需要确保释放后再执行后续操作，用 unpersist(blocking=true)。

练习

  1. 对同一个 RDD 分别用 MEMORY_ONLY 和 MEMORY_ONLY_SER 缓存，在 Spark UI 的 Storage 页面对比两者的 Size in Memory。序列化存储通常占用更少内存，但计算时需要反序列化。

  2. 创建一个广播变量（如一个 100MB 的 Map），在有 4 个 Executor 的集群上执行作业。在 Spark UI 的 Executors 页面观察每个 Executor 的 Storage Memory 增量是否都是约 100MB。

  3. 在一个 map transformation 中使用累加器，然后对同一个 RDD 调用两次不同的 action。观察累加器的值是否是预期的两倍。思考为什么 Spark 官方文档建议只在 action 中使用累加器。

系列导航

序号 主题 状态
00 导读：为什么 Spark 的核心是一张 DAG 已完成
01 RDD：弹性分布式数据集的五大属性 已完成
02 宽依赖与窄依赖：Stage 是怎样划分出来的 已完成
03 DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行 已完成
04 Shuffle 机制：数据跨分区交换的代价与优化 已完成
05 内存管理与 Tungsten：堆外内存、序列化与代码生成 已完成
06 存储体系：Block Manager、广播变量与累加器 本篇
07 容错机制：Lineage、Checkpoint 与推测执行 下一篇
08 Spark SQL 与 Catalyst：从 SQL 文本到物理算子树
09 DataFrame 与 Dataset：类型安全与性能的折中
10 Adaptive Query Execution：运行时改写执行计划
11 数据源 API 与谓词下推：让存储层帮忙过滤
12 Structured Streaming：把流当成无界表
13 水位线与状态管理：流处理的时间语义
14 Spark on YARN / K8s：资源管理与部署模式
15 动态资源分配与作业调度：多租户集群的资源博弈
16 性能诊断：Spark UI、Event Log 与调优思路
17 Spark 与 Flink：两种计算引擎的设计选择

参考资料