jerrypeng commented on code in PR #57286:
URL: https://github.com/apache/spark/pull/57286#discussion_r3600434793
##########
core/src/main/scala/org/apache/spark/SparkEnv.scala:
##########
@@ -323,28 +449,45 @@ class SparkEnv (
return
}
+ // The tracker is needed when the pipelined manager
(spark.shuffle.manager.incremental) is a
+ // StreamingShuffleManager -- which is the default -- to serve pipelined
shuffle dependencies.
+ // Inspect the already-instantiated manager rather than re-reading the
config, so an empty
+ // (disabled) incremental slot is correctly treated as "no tracker".
+ val incrementalIsStreaming =
+ _pipelinedShuffleManager.exists(_.isInstanceOf[StreamingShuffleManager])
+ if (incrementalIsStreaming) {
+ createStreamingShuffleOutputTracker()
+ return
+ }
+
+ // 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.
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)
- }
+ if (shuffleManagerName == classOf[MultiShuffleManager].getName) {
Review Comment:
Can we consolidate the logic with incrementalIsStreaming check?
--
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]
