cloud-fan commented on code in PR #57286:
URL: https://github.com/apache/spark/pull/57286#discussion_r3601063127
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -323,28 +450,43 @@ class SparkEnv (
return
}
- val shuffleManagerName = ShuffleManager.getShuffleManagerClassName(conf)
- if (shuffleManagerName == classOf[StreamingShuffleManager].getName
- || shuffleManagerName == classOf[MultiShuffleManager].getName) {
- val tracker = if (SparkContext.isDriver(executorId)) {
- new StreamingShuffleOutputTrackerMaster(conf)
- } else {
- new StreamingShuffleOutputTrackerWorker(conf)
- }
+ // The tracker is needed when the pipelined manager
(spark.shuffle.manager.incremental) is a
+ // StreamingShuffleManager -- which is the default. Inspect the
already-instantiated manager
+ // rather than re-reading the config; this runs at the end of
initializeShuffleManager, so the
+ // manager is non-null here.
+ val incrementalIsStreaming =
+ _pipelinedShuffleManager.isInstanceOf[StreamingShuffleManager]
+ // It is also needed when a MultiShuffleManager is the blocking manager
(spark.shuffle.manager):
+ // it internally routes some shuffles to streaming. A bare
StreamingShuffleManager cannot be the
+ // blocking manager -- it is pipelined and rejected from that slot in
initializeShuffleManager.
+ // TODO: remove this MultiShuffleManager clause once MultiShuffleManager
is removed and
+ // streaming shuffles are served only through the incremental (pipelined)
manager slot.
+ val blockingIsMulti =
+ ShuffleManager.getShuffleManagerClassName(conf) ==
classOf[MultiShuffleManager].getName
Review Comment:
This inspects the config class name while the sibling
`incrementalIsStreaming` above inspects the already-instantiated manager.
`_blockingShuffleManager` exists at this point too, so inspecting the instance
is consistent and matches the object actually created (this is the
consolidation you asked about at line 467):
```suggestion
val blockingIsMulti =
_blockingShuffleManager.isInstanceOf[MultiShuffleManager]
```
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -76,15 +77,105 @@ class SparkEnv (
val outputCommitCoordinator: OutputCommitCoordinator,
val conf: SparkConf) extends Logging {
- // We initialize the ShuffleManager later in SparkContext and Executor to
allow
- // user jars to define custom ShuffleManagers.
- @volatile private var _shuffleManager: ShuffleManager = _
+ // The two shuffle managers are peers keyed by kind, not a default and an
override: a shuffle is
+ // routed to one or the other by its dependency type via
`shuffleManagerFor`, so neither is ever
+ // installed "behind" the other.
+ //
+ // The blocking manager (spark.shuffle.manager) serves all regular,
materialized shuffles and owns
+ // block-by-id resolution. It is always a BlockingShuffleManager --
initializeShuffleManager
+ // rejects a non-blocking manager in this slot -- so block-resolution has a
single, well-typed
+ // source. We initialize it later in SparkContext and Executor to allow user
jars to define custom
+ // ShuffleManagers.
+ @volatile private var _blockingShuffleManager: BlockingShuffleManager = _
+
+ // The pipelined manager (spark.shuffle.manager.incremental) serves pipelined
+ // (incrementally-readable) shuffle dependencies. It defaults to the built-in
+ // StreamingShuffleManager -- just as the blocking manager defaults to sort
-- and is always a
+ // PipelinedShuffleManager (initializeShuffleManager rejects a blocking
manager in this slot), so
+ // its output is never expected to be reachable through the block-manager
resolver. Like the
+ // blocking manager, it is initialized later in SparkContext and Executor to
allow user jars.
+ @volatile private var _pipelinedShuffleManager: PipelinedShuffleManager = _
// Latch to signal when the ShuffleManager has been initialized.
// Used to allow callers to wait for initialization.
private val shuffleManagerInitLatch = new CountDownLatch(1)
- def shuffleManager: ShuffleManager = _shuffleManager
+ /**
+ * The [[BlockingShuffleManager]] (configured by spark.shuffle.manager),
which serves all regular,
+ * materialized shuffle dependencies and owns block-by-id resolution. Use
this when the intent is
+ * specifically the blocking manager -- e.g. inspecting its concrete type.
To serve a specific
+ * shuffle's reads/writes, use `shuffleManagerFor`, which routes by
dependency type; to resolve a
+ * block by id, use `shuffleBlockResolver`.
+ */
+ def blockingShuffleManager: BlockingShuffleManager = _blockingShuffleManager
Review Comment:
The description says the new `SparkEnv` routing accessors are
`private[spark]` and that the only externally visible API change is the
`shuffleManager` deprecation — but `blockingShuffleManager`,
`pipelinedShuffleManager`, `shuffleBlockResolver`, and `shuffleManagerFor` are
public here on a `@DeveloperApi` class, so they're new public API returning
`private[spark]` types. Every caller is in-tree under `org.apache.spark`, so
`private[spark]` compiles cleanly and matches the stated intent. Either tighten
these to `private[spark]` or fix the description.
##########
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala:
##########
@@ -441,8 +435,12 @@ class BlockManagerMasterEndpoint(
mapStatuses.filter(_ != null).foreach { mapStatus =>
// Check if the executor has been deallocated
if
(!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
- val blocksToDel =
-
shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId,
mapStatus.mapId)
+ // Only a BlockingShuffleManager serves block-manager blocks; a
pipelined shuffle is
+ // served out-of-band and is not in the MapOutputTracker this
loop iterates, so this
+ // resolves only regular shuffles (None when the default manager
is not blocking).
Review Comment:
The parenthetical describes a state that can't occur:
`initializeShuffleManager` rejects a non-blocking manager in the default slot,
so the blocking manager is always blocking. `shuffleBlockResolver` is `None`
only before init.
```suggestion
// resolves only regular shuffles (None only before the
manager is initialized).
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
