Storage

Every JVM in a Spark cluster - driver and executors - hosts a BlockManager. It is the local block store and the per-node coordinator for cached RDDs, shuffle output, broadcast variables, and streaming state.

Block model

A block is the unit of storage. Each block is named by a BlockId (core/.../storage/BlockId.scala). The subtypes are stable across Spark versions:

BlockId subtype	Used for
RDDBlockId(rddId, partitionId)	Cached RDD partitions.
ShuffleBlockId(shuffleId, mapId, reduceId)	Sort-based shuffle output blocks.
ShuffleDataBlockId / ShuffleIndexBlockId	The data and index files of a single mapper.
ShufflePushBlockId / ShuffleMergedBlockId	Push-based shuffle blocks.
BroadcastBlockId(broadcastId, fieldName)	Pieces of a broadcast variable.
TaskResultBlockId(taskId)	Big task results that don't fit in the task RPC response.
StreamBlockId(streamId, uniqueId)	DStream and Structured Streaming intermediate blocks.
LogBlockId	Used by structured logging block writer (Spark 4).

A block lives in any combination of:

• Memory (deserialized or serialized).
• Disk.
• Off-heap (when off-heap is enabled).

Block manager architecture

graph TD
    subgraph Driver
        BMM[BlockManagerMaster]
        EP[BlockManagerMasterEndpoint]
        BMM --- EP
    end

    subgraph "Executor 1"
        BM1[BlockManager]
        MS1[MemoryStore]
        DS1[DiskStore]
        BM1 --- MS1
        BM1 --- DS1
    end

    subgraph "Executor 2"
        BM2[BlockManager]
        MS2[MemoryStore]
        DS2[DiskStore]
        BM2 --- MS2
        BM2 --- DS2
    end

    BM1 -.heartbeat / register block.- EP
    BM2 -.heartbeat / register block.- EP
    BM1 <-->|peer block fetch / replication| BM2
    BM1 -->|getBlockLocations| EP

Component	Source
BlockManager	core/.../storage/BlockManager.scala (~95 KB)
BlockManagerMaster (driver-side proxy)	core/.../storage/BlockManagerMaster.scala
BlockManagerMasterEndpoint (driver-side coordinator)	core/.../storage/BlockManagerMasterEndpoint.scala (~50 KB)
MemoryStore	core/.../storage/memory/MemoryStore.scala
DiskStore	core/.../storage/DiskStore.scala
DiskBlockManager	core/.../storage/DiskBlockManager.scala
BlockInfoManager (per-block read/write locks)	core/.../storage/BlockInfoManager.scala
ShuffleBlockFetcherIterator	core/.../storage/ShuffleBlockFetcherIterator.scala (~77 KB)
BlockReplicationPolicy	core/.../storage/BlockReplicationPolicy.scala

Storage operations

The main entry points on BlockManager:

• putSingle / putIterator / putBytes - store a block, optionally replicate.
• get / getLocalValues / getRemoteBytes - fetch a block, possibly from a peer.
• removeBlock / removeRdd - eviction / cleanup.
• releaseLock - release the BlockInfoManager lock.

Internally each call goes through BlockInfoManager, which gives the caller a read or write lock on the block id, and then through MemoryStore and/or DiskStore.

Memory store and eviction

MemoryStore keeps blocks in a LinkedHashMap. When new storage memory is requested:

1. The UnifiedMemoryManager (core/.../memory/UnifiedMemoryManager.scala) decides whether to grow storage, evict execution memory, or fail.
2. MemoryStore.evictBlocksToFreeSpace walks the LRU list and either drops or spills blocks to DiskStore.

The store can hold deserialized JVM objects (cheap to read, expensive in heap) or serialized bytes (cheaper in heap, costly to read).

Disk store

DiskStore writes blocks to one of N "local dirs" (spark.local.dir). DiskBlockManager spreads block files across the local dirs to balance disk IO. Files use deterministic names derived from the BlockId (spark-<rand>-temp_shuffle_* for shuffle, rdd_<id>_<part> for cache, etc.).

Decommissioning and fallback storage

BlockManagerDecommissioner (core/.../storage/BlockManagerDecommissioner.scala) gracefully migrates blocks off an executor that is shutting down. It first tries peer executors, then falls back to a configured FallbackStorage URI (core/.../storage/FallbackStorage.scala) - typically S3, ABFS, or HDFS.

Replication

When StorageLevel requests replication, BlockReplicationPolicy picks replication - 1 peers. The default BasicBlockReplicationPolicy prefers peers on different hosts. TopologyMapper (core/.../storage/TopologyMapper.scala) lets a deployment plug in rack-awareness.

Shuffle blocks

ShuffleBlockFetcherIterator is the reducer-side fetcher. It:

• Issues parallel block fetches up to spark.reducer.maxBlocksInFlightPerAddress.
• Spills oversize batches to disk to bound off-heap memory.
• Integrates with push-based shuffle through PushBasedFetchHelper (core/.../storage/PushBasedFetchHelper.scala) so reducers can pick up merged blocks from mergers.

For shuffle internals, see shuffle.md.

Integration points

• The SQL CACHE TABLE and Dataset.cache() calls go through BlockManager.putIterator.
• Broadcast variables in core/.../broadcast/TorrentBroadcast.scala ship pieces as BroadcastBlockId blocks, with the BlockManager handling deserialization.
• Streaming state stores (sql/core/.../execution/streaming/state/) write to a configured filesystem, not the BlockManager.

Entry points for modification

• Add a new BlockId variant: extend BlockId and update BlockId.apply for the parser.
• Add a new replication policy: implement BlockReplicationPolicy and configure spark.storage.replication.policy.
• Add a new fallback storage backend: implement the Hadoop FileSystem interface (most cloud stores already do) and configure spark.storage.decommission.fallbackStorage.path.