[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r626222342
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -689,7 +716,7 @@ private[spark] class DAGScheduler(
dep match {
case shufDep: ShuffleDependency[_, _, _] =>
val mapStage = getOrCreateShuffleMapStage(shufDep,
stage.firstJobId)
-if (!mapStage.isAvailable) {
+if (!mapStage.isAvailable || !mapStage.isMergeFinalized) {
Review comment:
Comment is there as part of `isMergeFinalized` API similar to
`isAvailable`. Do you think that's not enough?
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691: URL: https://github.com/apache/spark/pull/30691#discussion_r626223337 ## File path: core/src/test/resources/META-INF/services/org.apache.spark.scheduler.ExternalClusterManager ## @@ -1,3 +1,4 @@ org.apache.spark.scheduler.DummyExternalClusterManager org.apache.spark.scheduler.MockExternalClusterManager org.apache.spark.scheduler.CSMockExternalClusterManager +org.apache.spark.scheduler.PushBasedClusterManager Review comment: Yes this is needed for tests. -- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r626224298
##
File path:
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
##
@@ -3393,6 +3406,271 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
assert(rprofsE === Set())
}
+ private def initPushBasedShuffleConfs(conf: SparkConf) = {
+conf.set(config.SHUFFLE_SERVICE_ENABLED, true)
+conf.set(config.PUSH_BASED_SHUFFLE_ENABLED, true)
+conf.set("spark.master", "pushbasedshuffleclustermanager")
+ }
+
+ test("SPARK-32920: shuffle merge finalization") {
+initPushBasedShuffleConfs(conf)
+DAGSchedulerSuite.clearMergerLocs
+DAGSchedulerSuite.addMergerLocs(Seq("host1", "host2", "host3", "host4",
"host5"))
+val parts = 2
+val shuffleMapRdd = new MyRDD(sc, parts, Nil)
+val shuffleDep = new ShuffleDependency(shuffleMapRdd, new
HashPartitioner(parts))
+val reduceRdd = new MyRDD(sc, parts, List(shuffleDep), tracker =
mapOutputTracker)
+
+// Submit a reduce job that depends which will create a map stage
+submit(reduceRdd, (0 until parts).toArray)
+completeShuffleMapStageSuccessfully(0, 0, parts)
+assert(mapOutputTracker.getNumAvailableMergeResults(shuffleDep.shuffleId)
== parts)
+completeNextResultStageWithSuccess(1, 0)
+assert(results === Map(0 -> 42, 1 -> 42))
+results.clear()
+assertDataStructuresEmpty()
+ }
+
+ test("SPARK-32920: merger locations not empty") {
+initPushBasedShuffleConfs(conf)
+conf.set(config.SHUFFLE_MERGER_LOCATIONS_MIN_STATIC_THRESHOLD, 3)
+DAGSchedulerSuite.clearMergerLocs
+DAGSchedulerSuite.addMergerLocs(Seq("host1", "host2", "host3", "host4",
"host5"))
+val parts = 2
+
+val shuffleMapRdd = new MyRDD(sc, parts, Nil)
+val shuffleDep = new ShuffleDependency(shuffleMapRdd, new
HashPartitioner(parts))
+val reduceRdd = new MyRDD(sc, parts, List(shuffleDep), tracker =
mapOutputTracker)
+
+// Submit a reduce job that depends which will create a map stage
+submit(reduceRdd, (0 until parts).toArray)
+completeShuffleMapStageSuccessfully(0, 0, parts)
+val shuffleStage =
scheduler.stageIdToStage(0).asInstanceOf[ShuffleMapStage]
+assert(shuffleStage.shuffleDep.getMergerLocs.nonEmpty)
+
+assert(mapOutputTracker.getNumAvailableMergeResults(shuffleDep.shuffleId)
== parts)
+completeNextResultStageWithSuccess(1, 0)
+assert(results === Map(0 -> 42, 1 -> 42))
+
+results.clear()
+assertDataStructuresEmpty()
+ }
+
+ test("SPARK-32920: merger locations reuse from shuffle dependency") {
+initPushBasedShuffleConfs(conf)
+conf.set(config.SHUFFLE_MERGER_MAX_RETAINED_LOCATIONS, 3)
+DAGSchedulerSuite.clearMergerLocs
+DAGSchedulerSuite.addMergerLocs(Seq("host1", "host2", "host3", "host4",
"host5"))
+val parts = 2
+
+val shuffleMapRdd = new MyRDD(sc, parts, Nil)
+val shuffleDep = new ShuffleDependency(shuffleMapRdd, new
HashPartitioner(parts))
+val reduceRdd = new MyRDD(sc, parts, List(shuffleDep), tracker =
mapOutputTracker)
+submit(reduceRdd, Array(0, 1))
+
+completeShuffleMapStageSuccessfully(0, 0, parts)
+assert(shuffleDep.getMergerLocs.nonEmpty)
+val mergerLocs = shuffleDep.getMergerLocs
+completeNextResultStageWithSuccess(1, 0 )
+
+// submit another job w/ the shared dependency, and have a fetch failure
+val reduce2 = new MyRDD(sc, 2, List(shuffleDep))
+submit(reduce2, Array(0, 1))
+// Note that the stage numbering here is only b/c the shared dependency
produces a new, skipped
+// stage. If instead it reused the existing stage, then this would be
stage 2
+completeNextStageWithFetchFailure(3, 0, shuffleDep)
+scheduler.resubmitFailedStages()
+
+assert(scheduler.runningStages.nonEmpty)
+assert(scheduler.stageIdToStage(2)
+ .asInstanceOf[ShuffleMapStage].shuffleDep.getMergerLocs.nonEmpty)
+val newMergerLocs = scheduler.stageIdToStage(2)
+ .asInstanceOf[ShuffleMapStage].shuffleDep.getMergerLocs
+
+// Check if same merger locs is reused for the new stage with shared
shuffle dependency
+assert(mergerLocs.zip(newMergerLocs).forall(x => x._1.host == x._2.host))
+completeShuffleMapStageSuccessfully(2, 0, 2)
+completeNextResultStageWithSuccess(3, 1, idx => idx + 1234)
+assert(results === Map(0 -> 1234, 1 -> 1235))
+
+assertDataStructuresEmpty()
+ }
+
+ test("SPARK-32920: Disable shuffle merge due to not enough mergers
available") {
+initPushBasedShuffleConfs(conf)
+conf.set(config.SHUFFLE_MERGER_LOCATIONS_MIN_STATIC_THRESHOLD, 6)
+DAGSchedulerSuite.clearMergerLocs
+DAGSchedulerSuite.addMergerLocs(Seq("host1", "host2", "host3", "host4",
"host5"))
+val parts = 7
+
+val shuffleMapRdd = new MyRDD(sc, parts, Nil)
+val shuffleDep = new ShuffleDependency(shuffleMapRdd, new
HashPartitioner(parts))
+val reduceRdd =
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691: URL: https://github.com/apache/spark/pull/30691#discussion_r626773776 ## File path: core/src/main/scala/org/apache/spark/Dependency.scala ## @@ -102,6 +102,12 @@ class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag]( */ private[spark] var mergerLocs: Seq[BlockManagerId] = Nil + /** + * Stores the information about whether the shuffle merge is finalized for the shuffle map stage + * associated with this shuffle dependency + */ + private[this] var shuffleMergedFinalized: Boolean = false Review comment: I think this is with in `ShuffleDependency` class only but `ShuffleDependency` class is in `Dependency.scala` file. -- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r626774584
##
File path: core/src/main/scala/org/apache/spark/internal/config/package.scala
##
@@ -2075,6 +2075,25 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val PUSH_BASED_SHUFFLE_MERGE_RESULTS_TIMEOUT =
+ConfigBuilder("spark.shuffle.push.merge.results.timeout")
+ .doc("Specify the max amount of time DAGScheduler waits for the merge
results from " +
+"all remote shuffle services for a given shuffle. DAGScheduler will
start to submit " +
+"following stages if not all results are received within the timeout.")
+ .version("3.1.0")
+ .stringConf
+ .createWithDefault("10s")
+
+ private[spark] val PUSH_BASED_SHUFFLE_MERGE_FINALIZE_TIMEOUT =
+ConfigBuilder("spark.shuffle.push.merge.finalize.timeout")
+ .doc("Specify the amount of time DAGScheduler waits after all mappers
finish for " +
+"a given shuffle map stage before it starts sending merge finalize
requests to " +
+"remote shuffle services. This allows the shuffle services some extra
time to " +
+"merge as many blocks as possible.")
+ .version("3.1.0")
+ .stringConf
+ .createWithDefault("10s")
+
Review comment:
I think currently this is hard to tune but once we have the changes in
for `SPARK-33701` which does adaptive merge finalization, mostly this should be
taken care of.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r628401105
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -254,6 +259,28 @@ private[spark] class DAGScheduler(
private val blockManagerMasterDriverHeartbeatTimeout =
sc.getConf.get(config.STORAGE_BLOCKMANAGER_MASTER_DRIVER_HEARTBEAT_TIMEOUT).millis
+ private val shuffleMergeResultsTimeoutSec =
+
JavaUtils.timeStringAsSec(sc.getConf.get(config.PUSH_BASED_SHUFFLE_MERGE_RESULTS_TIMEOUT))
+
+ private val shuffleMergeFinalizeWaitSec =
+
JavaUtils.timeStringAsSec(sc.getConf.get(config.PUSH_BASED_SHUFFLE_MERGE_FINALIZE_TIMEOUT))
+
+ // lazy initialized so that the shuffle client can be properly initialized
+ private lazy val externalShuffleClient: Option[ExternalBlockStoreClient] =
+if (pushBasedShuffleEnabled) {
+ val transConf = SparkTransportConf.fromSparkConf(sc.conf, "shuffle", 1)
+ val shuffleClient = new ExternalBlockStoreClient(transConf,
env.securityManager,
+env.securityManager.isAuthenticationEnabled(),
+sc.conf.get(config.SHUFFLE_REGISTRATION_TIMEOUT))
+ shuffleClient.init(sc.conf.getAppId)
+ Some(shuffleClient)
+} else {
+ None
+}
+
+ private val shuffleMergeFinalizeScheduler =
+
ThreadUtils.newDaemonSingleThreadScheduledExecutor("shuffle-merge-finalizer")
Review comment:
Changed it to use `newDaemonThreadPoolScheduledExecutor` with 8 threads.
Don't think we need this to be configurable, let me know your thoughts.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691: URL: https://github.com/apache/spark/pull/30691#discussion_r629844923 ## File path: .idea/vcs.xml ## @@ -1,36 +0,0 @@ - Review comment: Somehow my idea file got added and pushed. I think I removed it. Isn't it? Let me check again. -- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r629845232
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -1271,21 +1302,28 @@ private[spark] class DAGScheduler(
* locations for block push/merge by getting the historical locations of
past executors.
*/
private def prepareShuffleServicesForShuffleMapStage(stage:
ShuffleMapStage): Unit = {
-// TODO(SPARK-32920) Handle stage reuse/retry cases separately as without
finalize
-// TODO changes we cannot disable shuffle merge for the retry/reuse cases
-val mergerLocs = sc.schedulerBackend.getShufflePushMergerLocations(
- stage.shuffleDep.partitioner.numPartitions, stage.resourceProfileId)
-
-if (mergerLocs.nonEmpty) {
- stage.shuffleDep.setMergerLocs(mergerLocs)
- logInfo(s"Push-based shuffle enabled for $stage (${stage.name}) with" +
-s" ${stage.shuffleDep.getMergerLocs.size} merger locations")
-
- logDebug("List of shuffle push merger locations " +
-s"${stage.shuffleDep.getMergerLocs.map(_.host).mkString(", ")}")
-} else {
- logInfo("No available merger locations." +
-s" Push-based shuffle disabled for $stage (${stage.name})")
+if (stage.shuffleDep.shuffleMergeEnabled &&
!stage.shuffleDep.shuffleMergeFinalized
+ && stage.shuffleDep.getMergerLocs.isEmpty) {
+ val mergerLocs = sc.schedulerBackend.getShufflePushMergerLocations(
+stage.shuffleDep.partitioner.numPartitions, stage.resourceProfileId)
+ if (mergerLocs.nonEmpty) {
+stage.shuffleDep.setMergerLocs(mergerLocs)
+logInfo(s"Push-based shuffle enabled for $stage (${stage.name}) with" +
+ s" ${stage.shuffleDep.getMergerLocs.size} merger locations")
+
+logDebug("List of shuffle push merger locations " +
+ s"${stage.shuffleDep.getMergerLocs.map(_.host).mkString(", ")}")
+ } else {
+stage.shuffleDep.setShuffleMergeEnabled(false)
+logInfo("Push-based shuffle disabled for $stage (${stage.name})")
+ }
+} else if (stage.shuffleDep.shuffleMergeFinalized) {
+ // Disable Shuffle merge for the retry/reuse of the same shuffle
dependency if it has
+ // already been merge finalized. If the shuffle dependency was
previously assigned merger
+ // locations but the corresponding shuffle map stage did not complete
successfully, we
+ // would still enable push for its retry.
Review comment:
Yes, we are disabling merge in those case since it is already finalized.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691: URL: https://github.com/apache/spark/pull/30691#discussion_r629848900 ## File path: .idea/vcs.xml ## @@ -1,36 +0,0 @@ - Review comment: Sorry my bad, I think it got added as part of this `[SPARK-35223] Add IssueNavigationLink` I thought I added it by mistake. Fixed it. Should be good now. -- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r639104565
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -2004,6 +2020,131 @@ private[spark] class DAGScheduler(
}
}
+ /**
+ * Schedules shuffle merge finalize.
+ */
+ private[scheduler] def scheduleShuffleMergeFinalize(stage: ShuffleMapStage):
Unit = {
+logInfo(("%s (%s) scheduled for finalizing" +
+ " shuffle merge in %s s").format(stage, stage.name,
shuffleMergeFinalizeWaitSec))
+shuffleMergeFinalizeScheduler.schedule(
+ new Runnable {
+override def run(): Unit = finalizeShuffleMerge(stage)
+ },
+ shuffleMergeFinalizeWaitSec,
+ TimeUnit.SECONDS
+)
+ }
+
+ /**
+ * DAGScheduler notifies all the remote shuffle services chosen to serve
shuffle merge request for
+ * the given shuffle map stage to finalize the shuffle merge process for
this shuffle. This is
+ * invoked in a separate thread to reduce the impact on the DAGScheduler
main thread, as the
+ * scheduler might need to talk to 1000s of shuffle services to finalize
shuffle merge.
+ */
+ private[scheduler] def finalizeShuffleMerge(stage: ShuffleMapStage): Unit = {
+logInfo("%s (%s) finalizing the shuffle merge".format(stage, stage.name))
Review comment:
Addressed this comment after discussing with @mridulm offline. Mridul, I
tried adding test for the cancellation but the
`DAGSchedulerEventProcessLoopTester` is not async as it simply forwards the
event for immediate processing. Let me know if you have other ideas to test
this particular situation.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r640027450
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -2136,9 +2137,24 @@ private[spark] class DAGScheduler(
}
}
- private[scheduler] def handleShuffleMergeFinalized(stage: ShuffleMapStage):
Unit = {
-stage.shuffleDep.markShuffleMergeFinalized
-processShuffleMapStageCompletion(stage)
+ private[scheduler] def handleRegisterMergeStatuses(
+ stage: ShuffleMapStage,
+ mergeStatuses: Seq[(Int, MergeStatus)]): Unit = {
+// Register merge statuses if the stage is still running and shuffle merge
is not finalized yet.
+if (runningStages.contains(stage) &&
!stage.shuffleDep.shuffleMergeFinalized) {
+ mapOutputTracker.registerMergeResults(stage.shuffleDep.shuffleId,
mergeStatuses)
+}
+ }
+
+ private[scheduler] def handleShuffleMergeFinalized(
+ stage: ShuffleMapStage): Unit = {
+// Only update MapOutputTracker metadata if the stage is still active. i.e
not cancelled.
+if (runningStages.contains(stage)) {
+ stage.shuffleDep.markShuffleMergeFinalized()
+ processShuffleMapStageCompletion(stage)
+} else {
+ mapOutputTracker.unregisterAllMergeResult(stage.shuffleDep.shuffleId)
Review comment:
Stage will be active until the shuffle merge is finalized and then only
we are processing the map stage completion, isn't it? So if the stage is not
part of running stages and we still reach the handling shuffle merge finalize,
then we need to unregister the merge results, isn't it?
Can you think of a scenario where stage is not part of running stages and
still shuffle merge is finalized? - Ideally this should not happen.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r641094842
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -2136,9 +2137,24 @@ private[spark] class DAGScheduler(
}
}
- private[scheduler] def handleShuffleMergeFinalized(stage: ShuffleMapStage):
Unit = {
-stage.shuffleDep.markShuffleMergeFinalized
-processShuffleMapStageCompletion(stage)
+ private[scheduler] def handleRegisterMergeStatuses(
+ stage: ShuffleMapStage,
+ mergeStatuses: Seq[(Int, MergeStatus)]): Unit = {
+// Register merge statuses if the stage is still running and shuffle merge
is not finalized yet.
+if (runningStages.contains(stage) &&
!stage.shuffleDep.shuffleMergeFinalized) {
+ mapOutputTracker.registerMergeResults(stage.shuffleDep.shuffleId,
mergeStatuses)
+}
+ }
+
+ private[scheduler] def handleShuffleMergeFinalized(
+ stage: ShuffleMapStage): Unit = {
+// Only update MapOutputTracker metadata if the stage is still active. i.e
not cancelled.
+if (runningStages.contains(stage)) {
+ stage.shuffleDep.markShuffleMergeFinalized()
+ processShuffleMapStageCompletion(stage)
+} else {
+ mapOutputTracker.unregisterAllMergeResult(stage.shuffleDep.shuffleId)
Review comment:
Discussed offline with @mridulm and currently there are few corner cases
which needs to be carefully thought through before having this behavior.
Created a TODO and a corresponding follow up JIRA -
https://issues.apache.org/jira/browse/SPARK-35549
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r641095507
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -2004,6 +2020,131 @@ private[spark] class DAGScheduler(
}
}
+ /**
+ * Schedules shuffle merge finalize.
+ */
+ private[scheduler] def scheduleShuffleMergeFinalize(stage: ShuffleMapStage):
Unit = {
+logInfo(("%s (%s) scheduled for finalizing" +
+ " shuffle merge in %s s").format(stage, stage.name,
shuffleMergeFinalizeWaitSec))
+shuffleMergeFinalizeScheduler.schedule(
+ new Runnable {
+override def run(): Unit = finalizeShuffleMerge(stage)
+ },
+ shuffleMergeFinalizeWaitSec,
+ TimeUnit.SECONDS
+)
+ }
+
+ /**
+ * DAGScheduler notifies all the remote shuffle services chosen to serve
shuffle merge request for
+ * the given shuffle map stage to finalize the shuffle merge process for
this shuffle. This is
+ * invoked in a separate thread to reduce the impact on the DAGScheduler
main thread, as the
+ * scheduler might need to talk to 1000s of shuffle services to finalize
shuffle merge.
+ */
+ private[scheduler] def finalizeShuffleMerge(stage: ShuffleMapStage): Unit = {
+logInfo("%s (%s) finalizing the shuffle merge".format(stage, stage.name))
Review comment:
Added additional tests to handle the cases of stage cancellation,
barrier stage, late arrival of merge results etc.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r641707976
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -2000,6 +2023,147 @@ private[spark] class DAGScheduler(
}
}
+ /**
+ * Schedules shuffle merge finalize.
+ */
+ private[scheduler] def scheduleShuffleMergeFinalize(stage: ShuffleMapStage):
Unit = {
+logInfo(("%s (%s) scheduled for finalizing" +
+ " shuffle merge in %s s").format(stage, stage.name,
shuffleMergeFinalizeWaitSec))
+shuffleMergeFinalizeScheduler.schedule(
+ new Runnable {
+override def run(): Unit = finalizeShuffleMerge(stage)
+ },
+ shuffleMergeFinalizeWaitSec,
+ TimeUnit.SECONDS
+)
+ }
+
+ /**
+ * DAGScheduler notifies all the remote shuffle services chosen to serve
shuffle merge request for
+ * the given shuffle map stage to finalize the shuffle merge process for
this shuffle. This is
+ * invoked in a separate thread to reduce the impact on the DAGScheduler
main thread, as the
+ * scheduler might need to talk to 1000s of shuffle services to finalize
shuffle merge.
+ */
+ private[scheduler] def finalizeShuffleMerge(stage: ShuffleMapStage): Unit = {
+logInfo("%s (%s) finalizing the shuffle merge".format(stage, stage.name))
+externalShuffleClient.foreach { shuffleClient =>
+ val shuffleId = stage.shuffleDep.shuffleId
+ val numMergers = stage.shuffleDep.getMergerLocs.length
+ val numResponses = new AtomicInteger()
+ val results = (0 until numMergers).map(_ =>
SettableFuture.create[Boolean]())
+ val timedOut = new AtomicBoolean()
+
+ def increaseAndCheckResponseCount(): Unit = {
+if (numResponses.incrementAndGet() == numMergers) {
+ logInfo("%s (%s) shuffle merge finalized".format(stage, stage.name))
+ // Since this runs in the netty client thread and is outside of
DAGScheduler
+ // event loop, we only post ShuffleMergeFinalized event into the
event queue.
+ // The processing of this event should be done inside the event
loop, so it
+ // can safely modify scheduler's internal state.
+ eventProcessLoop.post(ShuffleMergeFinalized(stage))
+}
+ }
+
+ stage.shuffleDep.getMergerLocs.zipWithIndex.foreach {
+case (shuffleServiceLoc, index) =>
+ // Sends async request to shuffle service to finalize shuffle merge
on that host
+ // TODO: SPARK-35536: Cancel finalizeShuffleMerge if the stage is
cancelled
+ // TODO: during shuffleMergeFinalizeWaitSec
+ shuffleClient.finalizeShuffleMerge(shuffleServiceLoc.host,
+shuffleServiceLoc.port, shuffleId,
+new MergeFinalizerListener {
+ override def onShuffleMergeSuccess(statuses: MergeStatuses):
Unit = {
+assert(shuffleId == statuses.shuffleId)
+if (!timedOut.get()) {
+ eventProcessLoop.post(RegisterMergeStatuses(stage,
MergeStatus.
+convertMergeStatusesToMergeStatusArr(statuses,
shuffleServiceLoc)))
+ increaseAndCheckResponseCount()
+ results(index).set(true)
+}
+ }
+
+ override def onShuffleMergeFailure(e: Throwable): Unit = {
+if (!timedOut.get()) {
+ logWarning(s"Exception encountered when trying to finalize
shuffle " +
+s"merge on ${shuffleServiceLoc.host} for shuffle
$shuffleId", e)
+ increaseAndCheckResponseCount()
+ // Do not fail the future as this would cause dag scheduler
to prematurely
+ // give up on waiting for merge results from the remaining
shuffle services
+ // if one fails
+ results(index).set(false)
+}
+ }
+})
+ }
+ // DAGScheduler only waits for a limited amount of time for the merge
results.
+ // It will attempt to submit the next stage(s) irrespective of whether
merge results
+ // from all shuffle services are received or not.
+ // TODO: SPARK-33701: Instead of waiting for a constant amount of time
for finalization
+ // TODO: for all the stages, adaptively tune timeout for merge
finalization
+ try {
+Futures.allAsList(results: _*).get(shuffleMergeResultsTimeoutSec,
TimeUnit.SECONDS)
+ } catch {
+case _: TimeoutException =>
+ logInfo(s"Timed out on waiting for merge results from all " +
+s"$numMergers mergers for shuffle $shuffleId")
+ timedOut.set(true)
+ eventProcessLoop.post(ShuffleMergeFinalized(stage))
+ }
+}
+ }
+
+ private def processShuffleMapStageCompletion(shuffleStage: ShuffleMapStage):
Unit = {
+markStageAsFinished(shuffleStage)
+logInfo("looking for newly runnable stages")
+lo
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r643344276
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -1743,6 +1762,10 @@ private[spark] class DAGScheduler(
} else if (mapIndex != -1) {
// Mark the map whose fetch failed as broken in the map stage
mapOutputTracker.unregisterMapOutput(shuffleId, mapIndex,
bmAddress)
+if (mapStage.shuffleDep.shuffleMergeEnabled) {
+ mapOutputTracker.
+unregisterMergeResult(shuffleId, reduceId, bmAddress,
Option(mapIndex))
Review comment:
@otterc one question which I couldn't understand from the internal fix
is, should we pass `mapId` or `mapIndex` to the `unregisterMergeResult` here?
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r643574768
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -1743,6 +1762,10 @@ private[spark] class DAGScheduler(
} else if (mapIndex != -1) {
// Mark the map whose fetch failed as broken in the map stage
mapOutputTracker.unregisterMapOutput(shuffleId, mapIndex,
bmAddress)
+if (mapStage.shuffleDep.shuffleMergeEnabled) {
+ mapOutputTracker.
+unregisterMergeResult(shuffleId, reduceId, bmAddress,
Option(mapIndex))
Review comment:
Thanks. Added a comment to keep it clear as well.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r643661092
##
File path: core/src/main/scala/org/apache/spark/MapOutputTracker.scala
##
@@ -737,25 +737,26 @@ private[spark] class MapOutputTrackerMaster(
}
/**
- * Unregisters a merge result corresponding to the reduceId if present. If
the optional mapId
- * is specified, it will only unregister the merge result if the mapId is
part of that merge
+ * Unregisters a merge result corresponding to the reduceId if present. If
the optional mapIndex
+ * is specified, it will only unregister the merge result if the mapIndex is
part of that merge
* result.
*
* @param shuffleId the shuffleId.
* @param reduceId the reduceId.
* @param bmAddress block manager address.
- * @param mapId the optional mapId which should be checked to see it was
part of the merge
- * result.
+ * @param mapIndex the optional mapIndex which should be checked to see it
was part of the
+ * merge result.
*/
def unregisterMergeResult(
shuffleId: Int,
reduceId: Int,
bmAddress: BlockManagerId,
-mapId: Option[Int] = None) {
+mapIndex: Option[Int] = None) {
shuffleStatuses.get(shuffleId) match {
case Some(shuffleStatus) =>
val mergeStatus = shuffleStatus.mergeStatuses(reduceId)
-if (mergeStatus != null && (mapId.isEmpty ||
mergeStatus.tracker.contains(mapId.get))) {
+if (mergeStatus != null &&
+ (mapIndex.isEmpty || mergeStatus.tracker.contains(mapIndex.get))) {
shuffleStatus.removeMergeResult(reduceId, bmAddress)
Review comment:
[SPARK-32923](https://issues.apache.org/jira/browse/SPARK-32923) would
handle non deterministic stage retries right? Do you mean we should remove the
`mapOutputTracker.unregisterMergeResult` call in `DAGScheduler`? This change is
already added as part of
[SPARK-32921](https://issues.apache.org/jira/browse/SPARK-32921)
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r646974281
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -2000,6 +2025,149 @@ private[spark] class DAGScheduler(
}
}
+ /**
+ * Schedules shuffle merge finalize.
+ */
+ private[scheduler] def scheduleShuffleMergeFinalize(stage: ShuffleMapStage):
Unit = {
+logInfo(("%s (%s) scheduled for finalizing" +
+ " shuffle merge in %s s").format(stage, stage.name,
shuffleMergeFinalizeWaitSec))
+shuffleMergeFinalizeScheduler.schedule(
+ new Runnable {
+override def run(): Unit = finalizeShuffleMerge(stage)
+ },
+ shuffleMergeFinalizeWaitSec,
+ TimeUnit.SECONDS
+)
+ }
+
+ /**
+ * DAGScheduler notifies all the remote shuffle services chosen to serve
shuffle merge request for
+ * the given shuffle map stage to finalize the shuffle merge process for
this shuffle. This is
+ * invoked in a separate thread to reduce the impact on the DAGScheduler
main thread, as the
+ * scheduler might need to talk to 1000s of shuffle services to finalize
shuffle merge.
+ */
+ private[scheduler] def finalizeShuffleMerge(stage: ShuffleMapStage): Unit = {
+logInfo("%s (%s) finalizing the shuffle merge".format(stage, stage.name))
+externalShuffleClient.foreach { shuffleClient =>
+ val shuffleId = stage.shuffleDep.shuffleId
+ val numMergers = stage.shuffleDep.getMergerLocs.length
+ val numResponses = new AtomicInteger()
+ val results = (0 until numMergers).map(_ =>
SettableFuture.create[Boolean]())
+ val timedOut = new AtomicBoolean()
Review comment:
After having offline discussion, we simplified it further just by
posting `ShuffleMergeFinalized` only in the finally block that way it
guarantees the event itself would get posted only once.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r647758831
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -2000,6 +2023,133 @@ private[spark] class DAGScheduler(
}
}
+ /**
+ * Schedules shuffle merge finalize.
+ */
+ private[scheduler] def scheduleShuffleMergeFinalize(stage: ShuffleMapStage):
Unit = {
+logInfo(("%s (%s) scheduled for finalizing" +
+ " shuffle merge in %s s").format(stage, stage.name,
shuffleMergeFinalizeWaitSec))
+shuffleMergeFinalizeScheduler.schedule(
+ new Runnable {
+override def run(): Unit = finalizeShuffleMerge(stage)
+ },
+ shuffleMergeFinalizeWaitSec,
+ TimeUnit.SECONDS
+)
+ }
+
+ /**
+ * DAGScheduler notifies all the remote shuffle services chosen to serve
shuffle merge request for
+ * the given shuffle map stage to finalize the shuffle merge process for
this shuffle. This is
+ * invoked in a separate thread to reduce the impact on the DAGScheduler
main thread, as the
+ * scheduler might need to talk to 1000s of shuffle services to finalize
shuffle merge.
+ */
+ private[scheduler] def finalizeShuffleMerge(stage: ShuffleMapStage): Unit = {
+logInfo("%s (%s) finalizing the shuffle merge".format(stage, stage.name))
+externalShuffleClient.foreach { shuffleClient =>
+ val shuffleId = stage.shuffleDep.shuffleId
+ val numMergers = stage.shuffleDep.getMergerLocs.length
+ val numResponses = new AtomicInteger()
+ val results = (0 until numMergers).map(_ =>
SettableFuture.create[Boolean]())
+
+ stage.shuffleDep.getMergerLocs.zipWithIndex.foreach {
+case (shuffleServiceLoc, index) =>
+ // Sends async request to shuffle service to finalize shuffle merge
on that host
+ // TODO: SPARK-35536: Cancel finalizeShuffleMerge if the stage is
cancelled
+ // TODO: during shuffleMergeFinalizeWaitSec
+ shuffleClient.finalizeShuffleMerge(shuffleServiceLoc.host,
+shuffleServiceLoc.port, shuffleId,
+new MergeFinalizerListener {
+ override def onShuffleMergeSuccess(statuses: MergeStatuses):
Unit = {
+assert(shuffleId == statuses.shuffleId)
+eventProcessLoop.post(RegisterMergeStatuses(stage, MergeStatus.
+ convertMergeStatusesToMergeStatusArr(statuses,
shuffleServiceLoc)))
+numResponses.incrementAndGet()
+results(index).set(true)
+ }
+
+ override def onShuffleMergeFailure(e: Throwable): Unit = {
+logWarning(s"Exception encountered when trying to finalize
shuffle " +
+ s"merge on ${shuffleServiceLoc.host} for shuffle
$shuffleId", e)
+numResponses.incrementAndGet()
+// Do not fail the future as this would cause dag scheduler to
prematurely
+// give up on waiting for merge results from the remaining
shuffle services
+// if one fails
+results(index).set(false)
+ }
+})
+ }
+ // DAGScheduler only waits for a limited amount of time for the merge
results.
+ // It will attempt to submit the next stage(s) irrespective of whether
merge results
+ // from all shuffle services are received or not.
+ // TODO: SPARK-33701: Instead of waiting for a constant amount of time
for finalization
+ // TODO: for all the stages, adaptively tune timeout for merge
finalization
+ try {
+Futures.allAsList(results: _*).get(shuffleMergeResultsTimeoutSec,
TimeUnit.SECONDS)
+ } catch {
+case _: TimeoutException =>
+ logInfo(s"Timed out on waiting for merge results from all " +
+s"$numMergers mergers for shuffle $shuffleId")
+ } finally {
+eventProcessLoop.post(ShuffleMergeFinalized(stage))
+ }
+}
+ }
+
+ private def processShuffleMapStageCompletion(shuffleStage: ShuffleMapStage):
Unit = {
+markStageAsFinished(shuffleStage)
+logInfo("looking for newly runnable stages")
+logInfo("running: " + runningStages)
+logInfo("waiting: " + waitingStages)
+logInfo("failed: " + failedStages)
+
+// This call to increment the epoch may not be strictly necessary, but it
is retained
+// for now in order to minimize the changes in behavior from an earlier
version of the
+// code. This existing behavior of always incrementing the epoch following
any
+// successful shuffle map stage completion may have benefits by causing
unneeded
+// cached map outputs to be cleaned up earlier on executors. In the future
we can
+// consider removing this call, but this will require some extra
investigation.
+// See https://github.com/apache/spark/pull/17955/files#r117385673 for
more details.
+mapOutputTracker.incrementEpoch()
+
+clearCacheLocs()
+
+
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r647763338
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -254,6 +259,29 @@ private[spark] class DAGScheduler(
private val blockManagerMasterDriverHeartbeatTimeout =
sc.getConf.get(config.STORAGE_BLOCKMANAGER_MASTER_DRIVER_HEARTBEAT_TIMEOUT).millis
+ private val shuffleMergeResultsTimeoutSec =
+sc.getConf.get(config.PUSH_BASED_SHUFFLE_MERGE_RESULTS_TIMEOUT)
+
+ private val shuffleMergeFinalizeWaitSec =
+sc.getConf.get(config.PUSH_BASED_SHUFFLE_MERGE_FINALIZE_TIMEOUT)
+
+ // Since SparkEnv gets initialized after DAGScheduler, externalShuffleClient
needs to be
+ // initialized lazily
+ private lazy val externalShuffleClient: Option[ExternalBlockStoreClient] =
+if (pushBasedShuffleEnabled) {
+ val transConf = SparkTransportConf.fromSparkConf(sc.conf, "shuffle", 1)
+ val shuffleClient = new ExternalBlockStoreClient(transConf,
env.securityManager,
+env.securityManager.isAuthenticationEnabled(),
+sc.conf.get(config.SHUFFLE_REGISTRATION_TIMEOUT))
+ shuffleClient.init(sc.conf.getAppId)
+ Some(shuffleClient)
+} else {
+ None
+}
+
+ private val shuffleMergeFinalizeScheduler =
+
ThreadUtils.newDaemonThreadPoolScheduledExecutor("shuffle-merge-finalizer", 8)
Review comment:
Yes, I don't think we need a config as of now. In future, if needed we
can always add a config.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r647868344
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -1298,7 +1329,7 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
-mapOutputTracker.unregisterAllMapOutput(sms.shuffleDep.shuffleId)
+
mapOutputTracker.unregisterAllMapAndMergeOutput(sms.shuffleDep.shuffleId)
Review comment:
Made a comment here to handle clean up of shuffle merge metadata as part
of handling SPARK-32923 (non-deterministic stage retries) and in SPARK-35547
(handling barrier execution mode)
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r647868622
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -1739,18 +1756,24 @@ private[spark] class DAGScheduler(
if (mapStage.rdd.isBarrier()) {
// Mark all the map as broken in the map stage, to ensure retry
all the tasks on
// resubmitted stage attempt.
-mapOutputTracker.unregisterAllMapOutput(shuffleId)
+mapOutputTracker.unregisterAllMapAndMergeOutput(shuffleId)
} else if (mapIndex != -1) {
// Mark the map whose fetch failed as broken in the map stage
mapOutputTracker.unregisterMapOutput(shuffleId, mapIndex,
bmAddress)
+if (mapStage.shuffleDep.shuffleMergeEnabled) {
Review comment:
Instead of checking for shuffle dependency flag which can be set to
false later only checking for the global `pushBasedShuffleEnabled` variable.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r647868756
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -1678,38 +1717,16 @@ private[spark] class DAGScheduler(
}
if (runningStages.contains(shuffleStage) &&
shuffleStage.pendingPartitions.isEmpty) {
- markStageAsFinished(shuffleStage)
- logInfo("looking for newly runnable stages")
- logInfo("running: " + runningStages)
- logInfo("waiting: " + waitingStages)
- logInfo("failed: " + failedStages)
-
- // This call to increment the epoch may not be strictly
necessary, but it is retained
- // for now in order to minimize the changes in behavior from an
earlier version of the
- // code. This existing behavior of always incrementing the epoch
following any
- // successful shuffle map stage completion may have benefits by
causing unneeded
- // cached map outputs to be cleaned up earlier on executors. In
the future we can
- // consider removing this call, but this will require some extra
investigation.
- // See
https://github.com/apache/spark/pull/17955/files#r117385673 for more details.
- mapOutputTracker.incrementEpoch()
-
- clearCacheLocs()
-
- if (!shuffleStage.isAvailable) {
-// Some tasks had failed; let's resubmit this shuffleStage.
-// TODO: Lower-level scheduler should also deal with this
-logInfo("Resubmitting " + shuffleStage + " (" +
shuffleStage.name +
- ") because some of its tasks had failed: " +
- shuffleStage.findMissingPartitions().mkString(", "))
-submitStage(shuffleStage)
+ if (!shuffleStage.isMergeFinalized &&
Review comment:
Added the check of `shuffleStage.isAvailable` here.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r647868756
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -1678,38 +1717,16 @@ private[spark] class DAGScheduler(
}
if (runningStages.contains(shuffleStage) &&
shuffleStage.pendingPartitions.isEmpty) {
- markStageAsFinished(shuffleStage)
- logInfo("looking for newly runnable stages")
- logInfo("running: " + runningStages)
- logInfo("waiting: " + waitingStages)
- logInfo("failed: " + failedStages)
-
- // This call to increment the epoch may not be strictly
necessary, but it is retained
- // for now in order to minimize the changes in behavior from an
earlier version of the
- // code. This existing behavior of always incrementing the epoch
following any
- // successful shuffle map stage completion may have benefits by
causing unneeded
- // cached map outputs to be cleaned up earlier on executors. In
the future we can
- // consider removing this call, but this will require some extra
investigation.
- // See
https://github.com/apache/spark/pull/17955/files#r117385673 for more details.
- mapOutputTracker.incrementEpoch()
-
- clearCacheLocs()
-
- if (!shuffleStage.isAvailable) {
-// Some tasks had failed; let's resubmit this shuffleStage.
-// TODO: Lower-level scheduler should also deal with this
-logInfo("Resubmitting " + shuffleStage + " (" +
shuffleStage.name +
- ") because some of its tasks had failed: " +
- shuffleStage.findMissingPartitions().mkString(", "))
-submitStage(shuffleStage)
+ if (!shuffleStage.isMergeFinalized &&
Review comment:
Added the check of `shuffleStage.isAvailable` here. Also consolidated
checking for merge finalized by getting rid of `isMergeFinalized` in
`ShuffleMapStage` by moving that into `ShuffleDependency`
`shuffleMergeFinalized`
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r647869229
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -2000,6 +2023,133 @@ private[spark] class DAGScheduler(
}
}
+ /**
+ * Schedules shuffle merge finalize.
+ */
+ private[scheduler] def scheduleShuffleMergeFinalize(stage: ShuffleMapStage):
Unit = {
+logInfo(("%s (%s) scheduled for finalizing" +
+ " shuffle merge in %s s").format(stage, stage.name,
shuffleMergeFinalizeWaitSec))
+shuffleMergeFinalizeScheduler.schedule(
+ new Runnable {
+override def run(): Unit = finalizeShuffleMerge(stage)
+ },
+ shuffleMergeFinalizeWaitSec,
+ TimeUnit.SECONDS
+)
+ }
+
+ /**
+ * DAGScheduler notifies all the remote shuffle services chosen to serve
shuffle merge request for
+ * the given shuffle map stage to finalize the shuffle merge process for
this shuffle. This is
+ * invoked in a separate thread to reduce the impact on the DAGScheduler
main thread, as the
+ * scheduler might need to talk to 1000s of shuffle services to finalize
shuffle merge.
+ */
+ private[scheduler] def finalizeShuffleMerge(stage: ShuffleMapStage): Unit = {
+logInfo("%s (%s) finalizing the shuffle merge".format(stage, stage.name))
+externalShuffleClient.foreach { shuffleClient =>
+ val shuffleId = stage.shuffleDep.shuffleId
+ val numMergers = stage.shuffleDep.getMergerLocs.length
+ val numResponses = new AtomicInteger()
+ val results = (0 until numMergers).map(_ =>
SettableFuture.create[Boolean]())
+
+ stage.shuffleDep.getMergerLocs.zipWithIndex.foreach {
+case (shuffleServiceLoc, index) =>
+ // Sends async request to shuffle service to finalize shuffle merge
on that host
+ // TODO: SPARK-35536: Cancel finalizeShuffleMerge if the stage is
cancelled
+ // TODO: during shuffleMergeFinalizeWaitSec
+ shuffleClient.finalizeShuffleMerge(shuffleServiceLoc.host,
+shuffleServiceLoc.port, shuffleId,
+new MergeFinalizerListener {
+ override def onShuffleMergeSuccess(statuses: MergeStatuses):
Unit = {
+assert(shuffleId == statuses.shuffleId)
+eventProcessLoop.post(RegisterMergeStatuses(stage, MergeStatus.
+ convertMergeStatusesToMergeStatusArr(statuses,
shuffleServiceLoc)))
+numResponses.incrementAndGet()
+results(index).set(true)
+ }
+
+ override def onShuffleMergeFailure(e: Throwable): Unit = {
+logWarning(s"Exception encountered when trying to finalize
shuffle " +
+ s"merge on ${shuffleServiceLoc.host} for shuffle
$shuffleId", e)
+numResponses.incrementAndGet()
+// Do not fail the future as this would cause dag scheduler to
prematurely
+// give up on waiting for merge results from the remaining
shuffle services
+// if one fails
+results(index).set(false)
+ }
+})
+ }
+ // DAGScheduler only waits for a limited amount of time for the merge
results.
+ // It will attempt to submit the next stage(s) irrespective of whether
merge results
+ // from all shuffle services are received or not.
+ // TODO: SPARK-33701: Instead of waiting for a constant amount of time
for finalization
+ // TODO: for all the stages, adaptively tune timeout for merge
finalization
Review comment:
Moved the comment to the right place.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] venkata91 commented on a change in pull request #30691: [SPARK-32920][SHUFFLE] Finalization of Shuffle push/merge with Push based shuffle and preparation step for the reduce stage
venkata91 commented on a change in pull request #30691:
URL: https://github.com/apache/spark/pull/30691#discussion_r647869895
##
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##
@@ -1739,18 +1756,24 @@ private[spark] class DAGScheduler(
if (mapStage.rdd.isBarrier()) {
// Mark all the map as broken in the map stage, to ensure retry
all the tasks on
// resubmitted stage attempt.
-mapOutputTracker.unregisterAllMapOutput(shuffleId)
+mapOutputTracker.unregisterAllMapAndMergeOutput(shuffleId)
Review comment:
Same as above, added a comment here to handle it as part of handling
barrier execution mode handling.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
