[GitHub] [spark] mridulm commented on a diff in pull request #40307: [SPARK-42689][CORE][SHUFFLE] Allow ShuffleDriverComponent to declare if shuffle data is reliably stored

["mridulm (via GitHub)" <gi...@apache.org>]

mridulm commented on code in PR #40307:
URL: https://github.com/apache/spark/pull/40307#discussion_r1130215354


##########
core/src/main/java/org/apache/spark/shuffle/api/ShuffleDriverComponents.java:
##########
@@ -61,4 +61,13 @@ default void registerShuffle(int shuffleId) {}
    * @param blocking Whether this call should block on the deletion of the data.
    */
   default void removeShuffle(int shuffleId, boolean blocking) {}
+
+  /**
+   * Does this shuffle component support reliable storage - external to the lifecycle of the
+   * executor host ? For example, writing shuffle data to a distributed filesystem or
+   * persisting it in a remote shuffle service.
+   */
+  default boolean supportsReliableStorage() {

Review Comment:
   In this context, we are looking to say the shuffle driver component `supports` reliable storage as a feature.



##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -103,7 +104,8 @@ private[spark] class ExecutorAllocationManager(
     conf: SparkConf,
     cleaner: Option[ContextCleaner] = None,
     clock: Clock = new SystemClock(),
-    resourceProfileManager: ResourceProfileManager)
+    resourceProfileManager: ResourceProfileManager,
+    shuffleDriverComponents: ShuffleDriverComponents)

Review Comment:
   Since `ExecutorAllocationManager` is `private[spark]`, if we need to pull more details in future - we can do that.
   Given this, I am making the change @dongjoon-hyun recommended - to pass the `boolean` instead of object: we can revisit it as this functionality evolves.



##########
core/src/test/scala/org/apache/spark/SparkContextSuite.scala:
##########
@@ -1399,6 +1399,14 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
     }
     sc.stop()
   }
+
+  test("ShuffleDataIO initialized after application id has been configured") {

Review Comment:
   Added jira id assuming that is what you meant.



##########
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala:
##########
@@ -985,6 +985,28 @@ class DAGSchedulerSuite extends SparkFunSuite with TempLocalSparkContext with Ti
     }
   }
 
+  test("worker lost with remote shuffle service enabled should not treat shuffle files lost") {

Review Comment:
   Added jira.



##########
core/src/main/scala/org/apache/spark/SparkContext.scala:
##########
@@ -596,6 +591,13 @@ class SparkContext(config: SparkConf) extends Logging {
       _conf.set(APP_ATTEMPT_ID, attemptId)
       _env.blockManager.blockStoreClient.setAppAttemptId(attemptId)
     }
+
+    // initialize after application id and attempt id has been initialized
+    _shuffleDriverComponents = ShuffleDataIOUtils.loadShuffleDataIO(_conf).driver()
+    _shuffleDriverComponents.initializeApplication().asScala.foreach { case (k, v) =>
+      _conf.set(ShuffleDataIOUtils.SHUFFLE_SPARK_CONF_PREFIX + k, v)
+    }

Review Comment:
   Good point, moved cleanup before cleanupApplication to be before `_heartbeater` stop



##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -203,7 +205,8 @@ private[spark] class ExecutorAllocationManager(
       throw new SparkException(
         s"s${DYN_ALLOCATION_SUSTAINED_SCHEDULER_BACKLOG_TIMEOUT.key} must be > 0!")
     }
-    if (!conf.get(config.SHUFFLE_SERVICE_ENABLED)) {
+    if (!conf.get(config.SHUFFLE_SERVICE_ENABLED) &&
+      !shuffleDriverComponents.supportsReliableStorage()) {

Review Comment:
   We can change that after SPARK-25299 support is complete.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org