You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "mridulm (via GitHub)" <gi...@apache.org> on 2024/07/22 08:39:28 UTC
[PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
mridulm opened a new pull request, #2639:
URL: https://github.com/apache/celeborn/pull/2639
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
Adds support for barrier stages.
This involves two aspects:
a) If there is a task failure when executing a barrier stage, all shuffle output for the stage attempt are discarded and ignored.
b) If there is a reexecution of a barrier stage (for ex, due to child stage getting a fetch failure), all shuffle output for the previous stage attempt are discarded and ignored.
This is similar to handling of indeterminate stages when `throwsFetchFailure` is `true`.
Note that this is supported only when `spark.celeborn.client.spark.fetch.throwsFetchFailure` is `true`
### Why are the changes needed?
As detailed in CELEBORN-1518, Celeborn currently does not support barrier stages; which is an essential functionality in Apache Spark which is widely in use by Spark users.
Enhancing Celeborn will allow its use for a wider set of Spark users.
### Does this PR introduce _any_ user-facing change?
Adds ability for Celeborn to support Apache Spark Barrier stages.
### How was this patch tested?
Existing tests, and additional tests (thanks to @jiang13021 in #2638)
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "jiang13021 (via GitHub)" <gi...@apache.org>.
jiang13021 commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694134039
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -855,7 +881,8 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
s"unexpected! unknown appShuffleId $appShuffleId when checking shuffle deterministic level"))
}
} else {
- shuffleIds.values.map(v => v._1).toSeq.reverse.find(isAllMaptaskEnd) match {
+ shuffleIds.values.filter(v => !isBarrierStage || v._2).map(v => v._1).toSeq.reverse.find(
Review Comment:
Why is there a check for `!isBarrierStage` here? It seems that there is no difference between the barrier stage and the indeterminate stage here. In other words, do we need to do a special check for the indeterminate stage?
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2254354114
Gentle ping on this @waitinfuture
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1713113839
##########
tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala:
##########
@@ -17,10 +17,13 @@
package org.apache.celeborn.tests.spark
-import java.io.File
+import java.io.{File, IOException}
+import java.util.concurrent.TimeUnit
Review Comment:
nit: unused import
##########
tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala:
##########
@@ -297,4 +300,165 @@ class CelebornFetchFailureSuite extends AnyFunSuite
sparkSession.stop()
}
}
+
+ test(s"celeborn spark integration test - resubmit an unordered barrier stage with throwsFetchFailure enabled") {
+ val sparkConf = new SparkConf().setAppName("rss-demo").setMaster("local[2]")
+ val sparkSession = SparkSession.builder()
+ .config(updateSparkConf(sparkConf, ShuffleMode.HASH))
+ .config("spark.sql.shuffle.partitions", 2)
+ .config("spark.celeborn.shuffle.forceFallback.partition.enabled", false)
+ .config("spark.celeborn.shuffle.enabled", "true")
Review Comment:
nit: seems this config is not defined by Celeborn?
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2259995540
+CC @RexXiong, @waitinfuture Please see the [fix here](https://github.com/apache/celeborn/pull/2639/commits/dcd8465dd98ee468de8579acf28a0234ce391fdd) for the determinate stage failure case we are discussing [here](https://github.com/apache/celeborn/pull/2609#discussion_r1697821776).
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "jiang13021 (via GitHub)" <gi...@apache.org>.
jiang13021 commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2255475682
LGTM. However, you might want to link to this PR in your comment: https://github.com/apache/celeborn/pull/2609. The PR you are referring to in the comments is a test I wrote for another issue.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
~~From a general flow You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` as well.~~
Thinking more, we need to do it before the stage attempt gets submitted, not after.
When spark does it, this gets handled properly - and so the shuffle epoch that gets propagated to tasks will be after the unregister has completed.
When Celeborn invokes `PbGetShuffleId`, when is where we could clear map status, that will be after the tasks have already started - and so results in updating the epoch post task start : resulting in mismatch.
This impacts how spark caches map status, among other things and gets very involved (even if we could fix it); and this would be subject to arbitrary change as spark evolves.
Given this, it would be better to rely on more well defined behaviors.
Note: I had prototyped removing this - when I realized the epoch issue !
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2260020199
Actually, let me think a bit more about the determinate stage case though ... there might be corner cases here that need handling.
Making the PR a draft.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
~~From a general flow You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` as well.~~
Thinking more, we need to do it before the stage attempt gets submitted, not after.
When spark does it, this gets handled properly - and so the shuffle epoch that gets propagated to tasks will be after the unregister has completed.
When Celeborn invokes `PbGetShuffleId`, which is when we could clear map status, that will be after the tasks have already started - and so results in updating the epoch post task start : resulting in mismatch.
The impact of epoch update is more involved and this would also be subject to arbitrary change as spark evolves.
Given this, it would be better to rely on more well defined behaviors.
Note: I had prototyped removing this - when I realized the epoch issue !
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
~~From a general flow You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` as well.~~
Thinking more, we need to do when before the stage attempt gets submitted, not after.
When spark does it, this gets handled properly - and so the shuffle epoch that gets propagated to tasks will be after the unregister has completed.
When Celeborn invokes `PbGetShuffleId`, and if we try to clear the map status, that will be after the tasks have already started - and so results in updating the epoch post task start : resulting in mismatch.
This will result in being a much more involved change - and while there might be a narrow path to ensuring it still works; this is subject to arbitrary change as spark evolves, and it would be better to rely on more well defined behaviors.
Note: I had prototyped removing this - when I realized the epoch issue !
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
~~From a general flow You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` as well.~~
Thinking more, we need to do when before the stage attempt gets submitted, not after.
When spark does it, this gets handled properly - and so the shuffle epoch that gets propagated to tasks will be after the unregister has completed.
When Celeborn invokes `PbGetShuffleId`, that is after the tasks have already started - and so results in updating the epoch post task start.
This will result in being a much more involved change - and while there might be a narrow patch to ensuring it still works; this is subject to arbitrary change, and it would be better to rely on more well defined behaviors.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694609764
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -855,7 +881,8 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
s"unexpected! unknown appShuffleId $appShuffleId when checking shuffle deterministic level"))
}
} else {
- shuffleIds.values.map(v => v._1).toSeq.reverse.find(isAllMaptaskEnd) match {
+ shuffleIds.values.filter(v => !isBarrierStage || v._2).map(v => v._1).toSeq.reverse.find(
Review Comment:
I agree, this should be fine - though I was minimizing the imapct of the change.
Updated to incorporate this change - thanks @jiang13021 !
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2261760819
I have reverted the changes for determinate stage handling here @RexXiong, @waitinfuture - and limiting it to barrier stage alone.
Let us handle task reattempt for determinate stages in #2609.
@jiang13021 , this is back to the same state you had reviewed and approved earlier at.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2283182118
Thanks! Merging to main(v0.6.0)/branch-0.5(v0.5.2)
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2282793142
Gentle ping @waitinfuture, @RexXiong
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "jiang13021 (via GitHub)" <gi...@apache.org>.
jiang13021 commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694116102
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
IMO, the `handleReportBarrierStageAttemptFailure` function serves two purposes:
1. unregisterMapOutput
2. unregisterShuffle
However, DAGScheduler in spark will unregister all mapOutput before resubmitting the barrier stage. Can we unregister the previous shuffle when generating a new shuffleId for the barrier stage and remove `handleReportBarrierStageAttemptFailure`?
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
~~From a general flow You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` as well.~~
Thinking more, we need to do it before the stage attempt gets submitted, not after.
When spark does it, this gets handled properly - and so the shuffle epoch that gets propagated to tasks will be after the unregister has completed.
When Celeborn invokes `PbGetShuffleId`, when is where we could clear map status, that will be after the tasks have already started - and so results in updating the epoch post task start : resulting in mismatch.
This impacts how spark caches map status, among other things and gets very involved (even if we could 'fix' it); and this would be subject to arbitrary change as spark evolves.
Given this, it would be better to rely on more well defined behaviors.
Note: I had prototyped removing this - when I realized the epoch issue !
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
~~From a general flow You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` as well.~~
Thinking more, we need to do it before the stage attempt gets submitted, not after.
When spark does it, the shuffle epoch that gets propagated to tasks will be after the unregister has completed.
When Celeborn invokes `PbGetShuffleId`, which is when we could clear map status, that will be after the tasks have already started - and so results in updating the epoch post task start : resulting in mismatch.
The impact of epoch update is more involved and this would also be subject to arbitrary change as spark evolves.
Given this, it would be better to rely on more well defined behaviors.
Note: I had prototyped removing this - when I realized the epoch issue !
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2248329700
+CC @waitinfuture, @jiang13021
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1713197805
##########
tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala:
##########
@@ -297,4 +300,165 @@ class CelebornFetchFailureSuite extends AnyFunSuite
sparkSession.stop()
}
}
+
+ test(s"celeborn spark integration test - resubmit an unordered barrier stage with throwsFetchFailure enabled") {
+ val sparkConf = new SparkConf().setAppName("rss-demo").setMaster("local[2]")
+ val sparkSession = SparkSession.builder()
+ .config(updateSparkConf(sparkConf, ShuffleMode.HASH))
+ .config("spark.sql.shuffle.partitions", 2)
+ .config("spark.celeborn.shuffle.forceFallback.partition.enabled", false)
+ .config("spark.celeborn.shuffle.enabled", "true")
Review Comment:
Indeed !
This was copy + paste of existing tests in the suite ... I should have paid more attention :-(
Let me remove this config from rest of the tests in this file as well. Thanks for catching it !
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` 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.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694609764
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -855,7 +881,8 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
s"unexpected! unknown appShuffleId $appShuffleId when checking shuffle deterministic level"))
}
} else {
- shuffleIds.values.map(v => v._1).toSeq.reverse.find(isAllMaptaskEnd) match {
+ shuffleIds.values.filter(v => !isBarrierStage || v._2).map(v => v._1).toSeq.reverse.find(
Review Comment:
I agree, this should be fine - though I was minimizing the imapct of the change.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2258928777
The test failure is not related to this change.
+CC @waitinfuture, @ErikFang
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2256709888
> However, you might want to link to this PR in your comment: https://github.com/apache/celeborn/pull/2609. The PR you are referring to in the comments is a test I wrote for another issue.
Thanks for pointing it out ! Updated
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
~~From a general flow You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` as well.~~
Thinking more, we need to do when before the stage attempt gets submitted, not after.
When spark does it, this gets handled properly - and so the shuffle epoch that gets propagated to tasks will be after the unregister has completed.
When Celeborn invokes `PbGetShuffleId`, and if we try to clear the map status, that will be after the tasks have already started - and so results in updating the epoch post task start : resulting in mismatch.
This will result in being a much more involved change - and while there might be a narrow patch to ensuring it still works; this is subject to arbitrary change, and it would be better to rely on more well defined behaviors.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1686196539
##########
client-spark/spark-2/src/main/scala/org/apache/spark/celeborn/ExceptionMakerHelper.scala:
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.celeborn
+
+import org.apache.spark.shuffle.FetchFailedException
+
+import org.apache.celeborn.common.util.ExceptionMaker
+
+object ExceptionMakerHelper {
Review Comment:
Pulled this `ExecutionMaker` out to help with the tests.
##########
client-spark/spark-2/src/main/scala/org/apache/spark/celeborn/ExceptionMakerHelper.scala:
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.celeborn
+
+import org.apache.spark.shuffle.FetchFailedException
+
+import org.apache.celeborn.common.util.ExceptionMaker
+
+object ExceptionMakerHelper {
Review Comment:
Pulled this `ExceptionMaker` out to help with the 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.
To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1686195241
##########
client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/BarrierHelper.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn;
+
+import org.apache.spark.BarrierTaskContext;
+import org.apache.spark.TaskContext;
+import org.apache.spark.shuffle.ShuffleHandle;
+
+import org.apache.celeborn.client.ShuffleClient;
+
+public class BarrierHelper {
+
+ public static void addFailureListenerIfBarrierTask(
+ ShuffleClient shuffleClient, TaskContext taskContext, ShuffleHandle handle) {
+
+ if (!(taskContext instanceof BarrierTaskContext)) return;
+ int appShuffleId = handle.shuffleId();
+ String appShuffleIdentifier = SparkUtils.getAppShuffleIdentifier(appShuffleId, taskContext);
Review Comment:
This class is essentially a duplication for 2.x and 3.x - since we need to access version specific `SparkUtils`
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1713199812
##########
tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala:
##########
@@ -297,4 +300,165 @@ class CelebornFetchFailureSuite extends AnyFunSuite
sparkSession.stop()
}
}
+
+ test(s"celeborn spark integration test - resubmit an unordered barrier stage with throwsFetchFailure enabled") {
+ val sparkConf = new SparkConf().setAppName("rss-demo").setMaster("local[2]")
+ val sparkSession = SparkSession.builder()
+ .config(updateSparkConf(sparkConf, ShuffleMode.HASH))
+ .config("spark.sql.shuffle.partitions", 2)
+ .config("spark.celeborn.shuffle.forceFallback.partition.enabled", false)
+ .config("spark.celeborn.shuffle.enabled", "true")
Review Comment:
Removed.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#issuecomment-2283169223
Thanks for the review @waitinfuture, have addressed comments.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture closed pull request #2639: [CELEBORN-1518] Add support for Apache Spark barrier stages
URL: https://github.com/apache/celeborn/pull/2639
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-1518] Add support for Apache Spark barrier stages [celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2639:
URL: https://github.com/apache/celeborn/pull/2639#discussion_r1694588299
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -913,6 +928,54 @@ class LifecycleManager(val appUniqueId: String, val conf: CelebornConf) extends
context.reply(pbReportShuffleFetchFailureResponse)
}
+ private def handleReportBarrierStageAttemptFailure(
+ context: RpcCallContext,
+ appShuffleId: Int,
+ appShuffleIdentifier: String): Unit = {
+
+ val shuffleIds = shuffleIdMapping.get(appShuffleId)
+ if (shuffleIds == null) {
+ throw new UnsupportedOperationException(s"unexpected! unknown appShuffleId $appShuffleId")
+ }
+ var ret = true
+ shuffleIds.synchronized {
+ shuffleIds.get(appShuffleIdentifier) match {
+ case Some((shuffleId, true)) =>
+ ret = invokeAppShuffleTrackerCallback(appShuffleId)
Review Comment:
~~From a general flow You are right, now that we are only supporting this when fetch failure is `throwsFetchFailure` enabled, we dont need it : it was necessary in the earlier iteration when `throwsFetchFailure` could be `false` as well.~~
Thinking more, we need to do when before the stage attempt gets submitted, not after.
When spark does it, this gets handled properly - and so the shuffle epoch that gets propagated to tasks will be after the unregister has completed.
When Celeborn invokes `PbGetShuffleId`, when is where we could clear map status, that will be after the tasks have already started - and so results in updating the epoch post task start : resulting in mismatch.
This impacts how spark caches map status, among other things and gets very involved (even if we could 'fix' it); and this would be subject to arbitrary change as spark evolves.
Given this, it would be better to rely on more well defined behaviors.
Note: I had prototyped removing this - when I realized the epoch issue !
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org