ExternalBlockHandler¶

ExternalBlockHandler is an RpcHandler.

Creating Instance¶

ExternalBlockHandler takes the following to be created:

• TransportConf
• Registered Executors File

ExternalBlockHandler creates the following:

• ShuffleMetrics
• OneForOneStreamManager
• ExternalShuffleBlockResolver

ExternalBlockHandler is created when:

• ExternalShuffleService is requested for an ExternalBlockHandler
• YarnShuffleService is requested to serviceInit

OneForOneStreamManager¶

ExternalBlockHandler can be given or creates an OneForOneStreamManager when created.

ExternalShuffleBlockResolver¶

ExternalBlockHandler can be given or creates an ExternalShuffleBlockResolver to be created.

ExternalShuffleBlockResolver is used for the following:

• registerExecutor when ExternalBlockHandler is requested to handle a RegisterExecutor message
• removeBlocks when ExternalBlockHandler is requested to handle a RemoveBlocks message
• getLocalDirs when ExternalBlockHandler is requested to handle a GetLocalDirsForExecutors message
• applicationRemoved when ExternalBlockHandler is requested to applicationRemoved
• executorRemoved when ExternalBlockHandler is requested to executorRemoved
• registerExecutor when ExternalBlockHandler is requested to reregisterExecutor

ExternalShuffleBlockResolver is used for the following:

• getBlockData and getRddBlockData for ManagedBufferIterator
• getBlockData and getContinuousBlocksData for ShuffleManagedBufferIterator

ExternalShuffleBlockResolver is closed when is ExternalBlockHandler.

Registered Executors File¶

ExternalBlockHandler can be given a Java's File (or null) to be created.

This file is simply to create an ExternalShuffleBlockResolver.

Messages¶

FetchShuffleBlocks¶

Request to read a set of blocks

"Posted" (created) when:

• OneForOneBlockFetcher is requested to createFetchShuffleBlocksMsg

When received, ExternalBlockHandler requests the OneForOneStreamManager to registerStream (with a ShuffleManagedBufferIterator).

ExternalBlockHandler prints out the following TRACE message to the logs:

Registered streamId [streamId] with [numBlockIds] buffers for client [clientId] from host [remoteAddress]

In the end, ExternalBlockHandler responds with a StreamHandle (of streamId and numBlockIds).

GetLocalDirsForExecutors¶

OpenBlocks¶

RegisterExecutor¶

RemoveBlocks¶

ShuffleMetrics¶

Executor Removed Notification¶

void executorRemoved(
  String executorId,
  String appId)

executorRemoved requests the ExternalShuffleBlockResolver to executorRemoved.

executorRemoved is used when:

• ExternalShuffleService is requested to executorRemoved

Application Finished Notification¶

void applicationRemoved(
  String appId,
  boolean cleanupLocalDirs)

applicationRemoved requests the ExternalShuffleBlockResolver to applicationRemoved.

applicationRemoved is used when:

• ExternalShuffleService is requested to applicationRemoved
• YarnShuffleService (Spark on YARN) is requested to stopApplication

Logging¶

Enable ALL logging level for org.apache.spark.network.shuffle.ExternalBlockHandler logger to see what happens inside.

Add the following line to conf/log4j.properties:

log4j.logger.org.apache.spark.network.shuffle.ExternalBlockHandler=ALL

Refer to Logging.