BlockManager — Key-Value Store of Blocks of Data

stop…​FIXME

getMatchingBlockIds…​FIXME

getLocalValues…​FIXME

Internally, when getLocalValues is executed, you should see the following DEBUG message in the logs:

getLocalValues obtains a read lock for blockId.

When no blockId block was found, you should see the following DEBUG message in the logs and getLocalValues returns “nothing” (i.e. NONE).

When the blockId block was found, you should see the following DEBUG message in the logs:

If blockId block has memory level and is registered in MemoryStore, getLocalValues returns a BlockResult as Memory read method and with a CompletionIterator for an interator:

getRemoteValues…​FIXME

get attempts to get the blockId block from a local block manager first before requesting it from remote block managers.

Internally, get tries to get the block from the local BlockManager. If the block was found, you should see the following INFO message in the logs and get returns the local BlockResult.

If however the block was not found locally, get tries to get the block from remote block managers. If retrieved from a remote block manager, you should see the following INFO message in the logs and get returns the remote BlockResult.

In the end, get returns “nothing” (i.e. NONE) when the blockId block was not found either in the local BlockManager or any remote BlockManager.

When the External Shuffle Service is enabled for a Spark application, BlockManager uses ExternalShuffleClient to read other executors’ shuffle files.

A Store is the place where blocks are held.

There are the following possible stores:

putBlockData simply stores blockId locally (given the given storage level).

Internally, putBlockData wraps ChunkedByteBuffer around data buffer’s NIO ByteBuffer and calls putBytes.

putBytes makes sure that the bytes are not null and doPutBytes.

doPut is an internal helper method for doPutBytes and doPutIterator.

doPut executes the input putBody function with a BlockInfo being a new BlockInfo object (with level storage level) that BlockInfoManager managed to create a write lock for.

If the block has already been created (and BlockInfoManager did not manage to create a write lock for), the following WARN message is printed out to the logs:

doPut releases the read lock for the block when keepReadLock flag is disabled and returns None immediately.

If however the write lock has been given, doPut executes putBody.

If the result of putBody is None the block is considered saved successfully.

For successful save and keepReadLock enabled, BlockInfoManager is requested to downgrade an exclusive write lock for blockId to a shared read lock.

For successful save and keepReadLock disabled, BlockInfoManager is requested to release lock on blockId.

For unsuccessful save, the block is removed from memory and disk stores and the following WARN message is printed out to the logs:

Ultimately, the following DEBUG message is printed out to the logs:

removeBlock removes the blockId block from the MemoryStore and DiskStore.

When executed, it prints out the following DEBUG message to the logs:

It requests BlockInfoManager for lock for writing for the blockId block. If it receives none, it prints out the following WARN message to the logs and quits.

Otherwise, with a write lock for the block, the block is removed from MemoryStore and DiskStore (see Removing Block in MemoryStore and Removing Block in DiskStore).

If both removals fail, it prints out the following WARN message:

The block is removed from BlockInfoManager.

It then calculates the current block status that is used to report the block status to the driver (if the input tellMaster and the info’s tellMaster are both enabled, i.e. true) and the current TaskContext metrics are updated with the change.

removeRdd removes all the blocks that belong to the rddId RDD.

It prints out the following INFO message to the logs:

It then requests RDD blocks from BlockInfoManager and removes them (from memory and disk) (without informing the driver).

The number of blocks removed is the final result.

removeBroadcast removes all the blocks of the input broadcastId broadcast.

Internally, it starts by printing out the following DEBUG message to the logs:

It then requests all the BroadcastBlockId objects that belong to the broadcastId broadcast from BlockInfoManager and removes them (from memory and disk).

The number of blocks removed is the final result.

BlockManager takes the following when created:

When created, BlockManager sets externalShuffleServiceEnabled internal flag per spark.shuffle.service.enabled Spark property.

BlockManager then creates an instance of DiskBlockManager (requesting deleteFilesOnStop when an external shuffle service is not in use).

BlockManager creates an instance of BlockInfoManager (as blockInfoManager).

BlockManager creates block-manager-future daemon cached thread pool with 128 threads maximum (as futureExecutionContext).

BlockManager creates a MemoryStore and DiskStore.

MemoryManager gets the MemoryStore object assigned.

BlockManager calculates the maximum memory to use (as maxMemory) by requesting the maximum on-heap and off-heap storage memory from the assigned MemoryManager.

BlockManager calculates the port used by the external shuffle service (as externalShuffleServicePort).

BlockManager creates a client to read other executors’ shuffle files (as shuffleClient). If the external shuffle service is used an ExternalShuffleClient is created or the input BlockTransferService is used.

BlockManager sets the maximum number of failures before this block manager refreshes the block locations from the driver (as maxFailuresBeforeLocationRefresh).

BlockManager registers BlockManagerSlaveEndpoint with the input RpcEnv, itself, and MapOutputTracker (as slaveEndpoint).

initialize initializes a BlockManager on the driver and executors (see Creating SparkContext Instance and Creating Executor Instance, respectively).

initialize does the following in order:

If the External Shuffle Service is used, the following INFO appears in the logs:

It registers itself to the driver’s BlockManagerMaster passing the BlockManagerId, the maximum memory (as maxMemory), and the BlockManagerSlaveEndpoint.

Ultimately, if the initialization happens on an executor and the External Shuffle Service is used, it registers to the shuffle service.

When executed, reregister prints the following INFO message to the logs:

reregister then registers itself to the driver’s BlockManagerMaster (just as it was when BlockManager was initializing). It passes the BlockManagerId, the maximum memory (as maxMemory), and the BlockManagerSlaveEndpoint.

reregister will then report all the local blocks to the BlockManagerMaster.

You should see the following INFO message in the logs:

For each block metadata (in BlockInfoManager) it gets block current status and tries to send it to the BlockManagerMaster.

If there is an issue communicating to the BlockManagerMaster, you should see the following ERROR message in the logs:

After the ERROR message, reregister stops reporting.

getCurrentBlockStatus returns the current BlockStatus of the BlockId block (with the block’s current StorageLevel, memory and disk sizes). It uses MemoryStore and DiskStore for size and other information.

Internally, it uses the input BlockInfo to know about the block’s storage level. If the storage level is not set (i.e. null), the returned BlockStatus assumes the default NONE storage level and the memory and disk sizes being 0.

If however the storage level is set, getCurrentBlockStatus uses MemoryStore and DiskStore to check whether the block is stored in the storages or not and request for their sizes in the storages respectively (using their getSize or assume 0).

reportAllBlocks…​FIXME

reportBlockStatus is an internal method for reporting a block status to the driver and if told to re-register it prints out the following INFO message to the logs:

It does asynchronous reregistration (using asyncReregister).

In either case, it prints out the following DEBUG message to the logs:

tryToReportBlockStatus reports block status update to BlockManagerMaster and returns its response.

When a new broadcast value is created, TorrentBroadcast blocks are put in the block manager.

You should see the following TRACE message:

It puts the data in the memory first and drop to disk if the memory store can’t hold it.

FIXME

block-manager-future is the execution context for…​FIXME

The underlying abstraction for blocks in Spark is a ByteBuffer that limits the size of a block to 2GB (Integer.MAX_VALUE – see Why does FileChannel.map take up to Integer.MAX_VALUE of data? and SPARK-1476 2GB limit in spark for blocks). This has implication not just for managed blocks in use, but also for shuffle blocks (memory mapped blocks are limited to 2GB, even though the API allows for long), ser-deser via byte array-backed output streams.

When a non-local executor starts, it initializes a BlockManager object using spark.app.id Spark property for the id.

BlockResult is a description of a fetched block with the readMethod and bytes.

registerTask registers the input taskAttemptId with BlockInfoManager.

getDiskWriter creates a DiskBlockObjectWriter with spark.shuffle.sync Spark property for syncWrites.

addUpdatedBlockStatusToTaskMetrics takes an active TaskContext (if available) and records updated BlockStatus for Block (in the task’s TaskMetrics).

shuffleMetricsSource requests the ShuffleClient for the shuffle-related metrics and creates a ShuffleMetricsSource with the source name per spark.shuffle.service.enabled configuration property:

replicate…​FIXME

replicateBlock…​FIXME

putIterator…​FIXME

putSingle…​FIXME

getRemoteBytes…​FIXME

getRemoteValues…​FIXME

getSingle…​FIXME

shuffleClient is a ShuffleClient that BlockManager uses for the following:

shuffleClient is also used when BlockStoreShuffleReader is requested to read combined key-value records for a reduce task (and creates a ShuffleBlockFetcherIterator).

shuffleClient can be either a ExternalShuffleClient or the BlockTransferService (that is the NettyBlockTransferService given by SparkEnv).

shuffleClient uses spark.shuffle.service.enabled configuration property (default: false) that controls whether to use ExternalShuffleClient (true) or the BlockTransferService (i.e. NettyBlockTransferService).

When created, BlockManager is given a BlockTransferService that is used for the following services:

The BlockTransferService acts as the ShuffleClient when spark.shuffle.service.enabled configuration property is off (which is the default).

When initialized, BlockManager requests the BlockTransferService to init. BlockManager also requests the ShuffleClient to init (that does nothing by default).

When stopped, BlockManager requests the BlockTransferService to close. BlockManager also requests the ShuffleClient to close.

getOrElseUpdate first attempts to get the block by the BlockId (from the local block manager first and, if unavailable, requesting remote peers).

getOrElseUpdate gives the BlockResult of the block if found.

If however the block was not found (in any block manager in a Spark cluster), getOrElseUpdate doPutIterator (for the input BlockId, the makeIterator function and the StorageLevel).

getOrElseUpdate branches off per the result.

For None, getOrElseUpdate getLocalValues for the BlockId and eventually returns the BlockResult (unless terminated by a SparkException due to some internal error).

For Some(iter), getOrElseUpdate returns an iterator of T values.

doPutIterator simply doPut with the putBody function that accepts a BlockInfo and does the following:

When dropFromMemory is executed, you should see the following INFO message in the logs:

It then asserts that the blockId block is locked for writing.

If the block’s StorageLevel uses disks and the internal DiskStore object (diskStore) does not contain the block, it is saved then. You should see the following INFO message in the logs:

The block’s memory size is fetched and recorded (using MemoryStore.getSize).

The block is removed from memory if exists. If not, you should see the following WARN message in the logs:

It then calculates the current storage status of the block and reports it to the driver. It only happens when info.tellMaster.

A block is considered updated when it was written to disk or removed from memory or both. If either happened, the current TaskContext metrics are updated with the change.

Ultimately, dropFromMemory returns the current storage level of the block.