SortShuffleManager — The Default (And Only) Sort-Based Shuffle System

SortShuffleManager is the one and only ShuffleManager in Spark with the short name sort or tungsten-sort.

Note
 You can use spark.shuffle.manager Spark property to activate your own implementation of ShuffleManager contract.
Caution
 FIXME The internal registries
Table 1. SortShuffleManager’s Internal Registries and Counters
Name Description

numMapsForShuffle

shuffleBlockResolver

IndexShuffleBlockResolver created when SortShuffleManager is created and used throughout the lifetime of the owning SortShuffleManager.

NOTE: shuffleBlockResolver is a part of ShuffleManager contract.

Beside the uses due to the contract, shuffleBlockResolver is used in unregisterShuffle and stopped in stop.
Tip

Enable DEBUG logging level for org.apache.spark.shuffle.sort.SortShuffleManager$ logger to see what happens inside.

Add the following line to conf/log4j.properties:

log4j.logger.org.apache.spark.shuffle.sort.SortShuffleManager$=DEBUG

Refer to Logging.

unregisterShuffle Method

Caution
 FIXME

Creating SortShuffleManager Instance

SortShuffleManager takes a SparkConf.

SortShuffleManager makes sure that spark.shuffle.spill Spark property is enabled. If not you should see the following WARN message in the logs:

WARN SortShuffleManager: spark.shuffle.spill was set to false, but this configuration is ignored as of Spark 1.6+. Shuffle will continue to spill to disk when necessary.

SortShuffleManager initializes the internal registries and counters.

Note
 SortShuffleManager is created when SparkEnv is created (on the driver and executors) which is at the very beginning of a Spark application’s lifecycle.

Creating ShuffleHandle (For ShuffleDependency) — registerShuffle Method

registerShuffle[K, V, C](
  shuffleId: Int,
  numMaps: Int,
  dependency: ShuffleDependency[K, V, C]): ShuffleHandle
Note
 registerShuffle is a part of ShuffleManager contract.
Caution
 FIXME Copy the conditions

registerShuffle returns a new ShuffleHandle that can be one of the following:

  1. BypassMergeSortShuffleHandle (with ShuffleDependency[K, V, V]) when shouldBypassMergeSort condition holds.

  2. SerializedShuffleHandle (with ShuffleDependency[K, V, V]) when canUseSerializedShuffle condition holds.

  3. BaseShuffleHandle

Selecting ShuffleWriter For ShuffleHandle — getWriter Method

getWriter[K, V](
  handle: ShuffleHandle,
  mapId: Int,
  context: TaskContext): ShuffleWriter[K, V]
Note
 getWriter is a part of ShuffleManager contract.

Internally, getWriter makes sure that a ShuffleHandle is associated with its numMaps in numMapsForShuffle internal registry.

Caution
 FIXME Associated?! What’s that?
Note
 getWriter expects that the input handle is of type BaseShuffleHandle (despite the signature that says that it can work with any ShuffleHandle). Moreover, getWriter further expects that in 2 (out of 3 cases) the input handle is a more specialized IndexShuffleBlockResolver.

getWriter then returns a new ShuffleWriter for the input ShuffleHandle:

  1. UnsafeShuffleWriter for SerializedShuffleHandle.

  2. BypassMergeSortShuffleWriter for BypassMergeSortShuffleHandle.

  3. SortShuffleWriter for BaseShuffleHandle.

Creating BlockStoreShuffleReader For ShuffleHandle — getReader Method

getReader[K, C](
  handle: ShuffleHandle,
  startPartition: Int,
  endPartition: Int,
  context: TaskContext): ShuffleReader[K, C]
Note
 getReader is a part of ShuffleManager contract.

getReader returns a new BlockStoreShuffleReader passing all the input parameters on to it.

Note
 getReader assumes that the input ShuffleHandle is of type BaseShuffleHandle.

Stopping SortShuffleManager — stop Method

stop(): Unit
Note
 stop is a part of ShuffleManager contract.

stop stops IndexShuffleBlockResolver (available as shuffleBlockResolver internal reference).

Considering BypassMergeSortShuffleHandle for ShuffleHandle — shouldBypassMergeSort Method

shouldBypassMergeSort(conf: SparkConf, dep: ShuffleDependency[_, _, _]): Boolean

shouldBypassMergeSort holds (i.e. is positive) when:

  1. The input ShuffleDependency has mapSideCombine flag enabled and aggregator defined.

  2. mapSideCombine flag is disabled (i.e. false) but the number of partitions (of the Partitioner of the input ShuffleDependency) is at most spark.shuffle.sort.bypassMergeThreshold Spark property (which defaults to 200).

Otherwise, shouldBypassMergeSort does not hold (i.e. false).

Note
 shouldBypassMergeSort is exclusively used when SortShuffleManager selects a ShuffleHandle (for a ShuffleDependency).

Considering SerializedShuffleHandle for ShuffleHandle — canUseSerializedShuffle Method

canUseSerializedShuffle(dependency: ShuffleDependency[_, _, _]): Boolean

canUseSerializedShuffle condition holds (i.e. is positive) when all of the following hold (checked in that order):

  1. The Serializer of the input ShuffleDependency supports relocation of serialized objects.

  2. The Aggregator of the input ShuffleDependency is not defined.

  3. The number of shuffle output partitions of the input ShuffleDependency is at most the supported maximum number (which is (1 << 24) - 1, i.e. 16777215).

You should see the following DEBUG message in the logs when canUseSerializedShuffle holds:

DEBUG Can use serialized shuffle for shuffle [id]

Otherwise, canUseSerializedShuffle does not hold and you should see one of the following DEBUG messages:

DEBUG Can't use serialized shuffle for shuffle [id] because the serializer, [name], does not support object relocation

DEBUG SortShuffleManager: Can't use serialized shuffle for shuffle [id] because an aggregator is defined

DEBUG Can't use serialized shuffle for shuffle [id] because it has more than [number] partitions
Note
 canUseSerializedShuffle is exclusively used when SortShuffleManager selects a ShuffleHandle (for a ShuffleDependency).

Settings

Table 2. Spark Properties
Spark Property Default Value Description

spark.shuffle.sort.bypassMergeThreshold

200

The maximum number of reduce partitions below which SortShuffleManager avoids merge-sorting data if there is no map-side aggregation either.

spark.shuffle.spill

true

No longer in use.

When false the following WARN shows in the logs when SortShuffleManager is created:

WARN SortShuffleManager: spark.shuffle.spill was set to false, but this configuration is ignored as of Spark 1.6+. Shuffle will continue to spill to disk when necessary.

results matching ""

    No results matching ""