Definition
- BlockManager – a key-value store running on the driver and every executor, storing and retrieving blocks (data partitions, broadcast variables, shuffle data) across on-heap memory, off-heap memory, and local disk.
- BlockManagerMaster – a central registry on the driver tracking which node holds which block, so any node can locate and fetch blocks.
- Four roles – caching (cached partitions), broadcasting (broadcast variables), shuffling (shuffle block exchange), and temporary storage (spill files).
- Locality – block fetches prefer local reads; remote blocks are fetched over the network.
Imagine every branch of the National Bank (each executor) has its own storage room (Block Manager) with shelves in fast memory, slower disk, and a special off-site vault. A head-office registry (BlockManagerMaster) records which branch's room holds which box of documents (block). When one branch needs a document another branch cached, it checks the registry, calls that branch, and fetches the box.
Understanding the Concepts
What is the Block Manager?
Every Spark node runs a BlockManager. It is a key-value store of blocks identified by a BlockId. It can place a block in on-heap memory, off-heap memory, or spill it to local disk depending on the storage level and available space.
from pyspark.sql import SparkSession
from pyspark import StorageLevel
import pyspark.sql.functions as F
spark = SparkSession.builder.appName("NBI-BlockManager").getOrCreate()
sc = spark.sparkContext
# The Block Manager is internal, but its effects are visible:
# Every cached partition, broadcast variable, and shuffle block
# is stored and served by the Block Manager.
How does the BlockManagerMaster track blocks?
The driver runs a BlockManagerMaster. Each executor's BlockManager registers with it and reports which blocks it holds. The master maintains a global map of block locations so any node can locate and fetch a block from the node that holds it.
# BlockManagerMaster (on driver) maintains:
# BlockId → [ExecutorId, StorageLevel, Size]
#
# Example entries:
# rdd_42_partition_0 → [executor_1, MEMORY_ONLY, 256 MB]
# broadcast_7 → [executor_1, MEMORY_AND_DISK, 50 MB]
# shuffle_3_map_0 → [executor_2, DISK_ONLY, 128 MB]
#
# When executor_3 needs rdd_42_partition_0:
# 1. Ask BlockManagerMaster: "Where is rdd_42_partition_0?"
# 2. Master responds: "executor_1"
# 3. Executor_3's BlockManager fetches from executor_1
How does the Block Manager handle caching?
When you call cache() or persist(), each partition becomes a block stored by the local BlockManager at the chosen storage level. On re-access, tasks read the cached blocks directly instead of recomputing.
txn = spark.read.parquet("hdfs:///bank/transactions/2026-01/")
# Cache with default MEMORY_AND_DISK
txn.cache()
txn.count() # materializes: each partition → a block in BlockManager
# Cache with explicit storage level
txn.persist(StorageLevel.MEMORY_AND_DISK_2) # 2 replicas
# Check cached partitions via Storage tab in Spark UI
# Each partition shows: size in memory, size on disk, storage level
# Unpersist: removes blocks from all executors' BlockManagers
txn.unpersist()
How does the Block Manager handle broadcasting?
When you broadcast a value, it is stored as a block on the driver's BlockManager. Executors fetch it once and keep it locally, so a large lookup table ships efficiently rather than being serialized into every task.
# Broadcast a lookup table
fee_rates = {"savings": 0.005, "current": 0.01, "loan": 0.02}
fee_bc = sc.broadcast(fee_rates)
# How it works:
# 1. fee_rates stored as a block on driver's BlockManager
# 2. First task on executor_1 requests the block
# 3. Executor_1's BlockManager fetches from driver, stores locally
# 4. Subsequent tasks on executor_1 read from local BlockManager
# Result: one network transfer per executor, not per task
print("Fee rates:", fee_bc.value)
Fee rates: {'savings': 0.005, 'current': 0.01, 'loan': 0.02}
How does the Block Manager handle shuffle data?
During a shuffle, map tasks write their output partitions as shuffle blocks through the BlockManager to local disk. Reduce tasks fetch the relevant shuffle blocks from the remote BlockManagers that produced them, using the master to locate them.
# Shuffle flow through Block Manager:
# Stage 1 (map side):
# Task on executor_1 writes shuffle blocks to local disk
# via executor_1's BlockManager
#
# Stage 2 (reduce side):
# Task on executor_3 needs shuffle data from executor_1
# 1. Asks BlockManagerMaster for location
# 2. Fetches shuffle block from executor_1's BlockManager
result = txn.groupBy("branch").agg(F.sum("amount"))
result.collect() # triggers shuffle: blocks exchanged via BlockManagers
How does the Block Manager handle spill files?
Intermediate spill files from sorts and aggregations that exceed execution memory are also managed through this storage layer.
# When a sort or aggregation exceeds execution memory:
# 1. Operator requests spill through BlockManager
# 2. BlockManager writes sorted/partitioned run to local disk
# 3. Memory is freed for more processing
# 4. Later: BlockManager reads spilled runs back for merge
# Heavy aggregation that may spill:
result = txn.groupBy("account_id", "branch") \
.agg(F.sum("amount"), F.count("*"), F.collect_list("txn_date")) \
.orderBy("branch")
# If execution memory is insufficient, BlockManager handles spill to disk
How does block locality work?
Block fetches prefer locality. A task reads a local block directly from the same executor's BlockManager and only goes over the network when the block lives on a different executor.
# Locality levels for block access:
# PROCESS_LOCAL: block is in the same executor's memory → direct read
# NODE_LOCAL: block is on the same machine's disk → local disk read
# RACK_LOCAL: block is in the same rack → rack-local network
# ANY: block is on a remote node → cross-network fetch
# Spark scheduler tries to place tasks where their blocks already are
# This minimizes network transfer for cached and shuffle data
How does the Block Manager interact with the unified memory model?
The Block Manager integrates with Spark's unified memory manager. Cached blocks and execution memory share and borrow space. Over-caching can starve execution and cause spills.
# Block Manager stores cached blocks in Storage Memory region
# Execution operators (shuffle, join, sort) use Execution Memory region
# Both share the unified pool
# Over-caching scenario:
txn.cache() # fills Storage Memory
# Then a heavy join needs Execution Memory
# Block Manager evicts cached blocks to make room
# If still not enough → execution spills to disk via Block Manager
# Fix: unpersist unused caches
txn.unpersist()
What should a senior engineer know?
A senior engineer understands that the Block Manager is the storage substrate for all four Spark subsystems: caching, broadcasting, shuffle, and temporary storage. They monitor the Storage tab in the Spark UI and know that over-caching competes with execution through the shared memory region.
# Senior knowledge:
# 1. Block Manager is per-node: driver + every executor
# 2. BlockManagerMaster on driver tracks all blocks globally
# 3. Four roles: cache, broadcast, shuffle, spill
# 4. Shares memory with execution via unified memory model
# 5. Monitor via Spark UI Storage tab
# 6. Block fetches prefer locality (PROCESS > NODE > RACK > ANY)
# 7. External shuffle service extends Block Manager's shuffle role
Visual Overview
Block Manager Architecture:
Driver Node Executor Nodes
┌──────────────────────┐ ┌───────────────────────┐
│ BlockManagerMaster │ │ Executor 1 │
│ (global block map) │ │ ┌─────────────────┐ │
│ │◄──────│ │ BlockManager │ │
│ rdd_42_p0 → exec_1 │ report│ │ ┌─────────────┐ │ │
│ broadcast_7 → driver │ │ │ │ Memory │ │ │
│ shuffle_3 → exec_2 │ │ │ │ (cached │ │ │
│ │ │ │ │ partitions)│ │ │
├───────────────────────┤ │ │ ├─────────────┤ │ │
│ Driver BlockManager │ │ │ │ Disk │ │ │
│ (broadcasts stored │ │ │ │ (shuffle, │ │ │
│ here first) │ │ │ │ spill) │ │ │
└──────────────────────┘ │ │ └─────────────┘ │ │
│ └─────────────────┘ │
└───────────────────────┘
Four Roles:
1. CACHING: cache()/persist() → blocks in memory/disk
2. BROADCAST: broadcast(val) → block on driver, fetched once per executor
3. SHUFFLE: map output → shuffle blocks on disk, fetched by reducers
4. TEMP STORAGE: spill files → overflow from sorts/aggs to disk
Key Takeaway
The Block Manager is Spark's distributed storage subsystem, and a BlockManager instance runs on the driver and on every executor. It is a key-value store of blocks, where a block is a unit of data like a cached partition, a broadcast variable, or shuffle output, and each BlockManager can place a block in on-heap memory, off-heap memory, or spill it to local disk depending on the storage level. On the driver there is also a BlockManagerMaster that every executor registers with, maintaining a global map of where each block lives.
It plays four roles I always name. For caching, when I call cache or persist, each partition becomes a block at the chosen storage level. For broadcasting, the broadcast value is stored as a block on the driver and fetched once per executor. For shuffling, map tasks write output as shuffle blocks to local disk, and reduce tasks locate and fetch remote shuffle blocks. And for temporary storage, spill files from sorts and aggregations are managed through this same layer.
At the National Bank of India this showed up constantly. I cached a cleaned transactions DataFrame so several reports reused the same blocks, I broadcast an account reference table so every join task read it locally, and the groupBy aggregates exchanged their intermediate data as shuffle blocks between executors.
The nuance I add is that block fetches prefer locality, and that the Block Manager shares memory with execution through the unified memory manager, which is why over-caching can starve execution and trigger spills.
Remember one sentence: the Block Manager is the per-node storage engine for cached partitions, broadcast variables, shuffle blocks, and spill files, with a master registry on the driver tracking where each block lives.
---