You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/01/22 09:05:35 UTC
[GitHub] [spark] weixiuli opened a new pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
weixiuli opened a new pull request #30716:
URL: https://github.com/apache/spark/pull/30716
### What changes were proposed in this pull request?
Avoid calling unregisterMapOutput when the map stage is being rerunning.
### Why are the changes needed?
When a fetch failure happened, DAGScheduler will try to unregister the corresponding map output. The current logic has a race condition that the new map stage attempt is running while the current reduce stage attempt returns another fetch failure (note: the current reduce stage firstly returns a fetch failure to make the maps stage is rerunning, and then the rerunning map stage may return some mapstatus of the failed MapId before the current reduce stage returns another fetch failure at the same MapId, the current reduce is last attempt due to the new map stage is not yet completed). In this case, if the map output is always unregistered, it may actually unregister the map output from the new map stage attempt.
### Does this PR introduce _any_ user-facing change?
No. It is a bug fix.
### How was this patch tested?
Add new uts
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-743935047
@Ngone51 PTAL, ping @dongjoon-hyun Kindly reivew.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on a change in pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on a change in pull request #30716:
URL: https://github.com/apache/spark/pull/30716#discussion_r546631296
##########
File path: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
##########
@@ -2035,6 +2040,107 @@ class DAGSchedulerSuite extends SparkFunSuite with TempLocalSparkContext with Ti
assert(scheduler.activeJobs.isEmpty)
}
+ def reInit(): Unit = {
+ assert(sc != null)
+ val dagOutputTracker = mapOutputTracker
+ val taskSchedulerImpl = new TaskSchedulerImpl(sc) {
+ override def submitTasks(taskSet: TaskSet) = {
+ super.submitTasks(taskSet)
+ taskSet.tasks.foreach(_.epoch = dagOutputTracker.getEpoch)
+ taskSets += taskSet
+ }
+
+ override def cancelTasks(stageId: Int, interruptThread: Boolean) {
+ cancelledStages += stageId
+ }
+ }
+ taskSchedulerImpl.initialize(new FakeSchedulerBackend)
+ scheduler = new DAGScheduler(
+ sc,
+ taskSchedulerImpl,
+ sc.listenerBus,
+ mapOutputTracker,
+ blockManagerMaster,
+ sc.env)
+ dagEventProcessLoopTester = new DAGSchedulerEventProcessLoopTester(scheduler)
+ }
+
+ test("Test dagScheduler.shouldUnregisterMapOutput with map stage not running") {
+ reInit()
+ val shuffleMapRdd = new MyRDD(sc, 2, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostB", reduceRdd.partitions.length))))
+ // The MapOutputTracker should know about both map output locations.
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId, 0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB"))
+
+ // The first result task fails, with a fetch failure for the output from the first mapper.
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0, 0, 0, "ignored"),
+ null))
+ assert(sparkListener.failedStages.contains(1))
+
+ val mapStatuses = mapOutputTracker.shuffleStatuses(shuffleId).mapStatuses
+ // unregisterMapOutput with a fetchFailed.
+ assert(mapStatuses.count(_ != null) === 1)
+ assert(mapStatuses(1).location === makeBlockManagerId("hostB"))
+ }
+
+ test("Test dagScheduler.shouldUnregisterMapOutput with map stage re-running") {
+ reInit()
+ val shuffleMapRdd = new MyRDD(sc, 3, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostB", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostC", reduceRdd.partitions.length))))
+ // The MapOutputTracker should know about both map output locations.
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId, 0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB", "hostC"))
+
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0, 0, 0, "ignored"),
+ null))
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(1),
+ FetchFailed(makeBlockManagerId("hostB"), shuffleId, 1, 1, 0, "ignored"),
+ null))
+
+ assert(sparkListener.failedStages.contains(1))
+
+ // Wait for a long time to make sure the map stage was resubmitted.
+ eventually(timeout(1000 milliseconds), interval(10 milliseconds)) {
+ assert(scheduler.runningStages.nonEmpty && taskSets.size == 3)
+ }
+
+ runEvent(makeCompletionEvent(
+ taskSets(2).tasks(0),
+ Success,
+ makeMapStatus("hostA", reduceRdd.partitions.length)))
+
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(1),
Review comment:
Oh, I'm sorry,i have updated my ut, thanks, PTAL.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli closed pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli closed pull request #30716:
URL: https://github.com/apache/spark/pull/30716
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
mridulm commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-747839969
I am a bit unclear on this - trying to understand what is the race condition here.
Particularly given [this issue in pr](https://github.com/apache/spark/pull/30716/files#diff-85de35b2e85646ed499c545a3be1cd3ffd525a88aae835a9c621f877eebadcb6R267), I am not sure what the actual behavior being tested here is.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on a change in pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on a change in pull request #30716:
URL: https://github.com/apache/spark/pull/30716#discussion_r546631296
##########
File path: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
##########
@@ -2035,6 +2040,107 @@ class DAGSchedulerSuite extends SparkFunSuite with TempLocalSparkContext with Ti
assert(scheduler.activeJobs.isEmpty)
}
+ def reInit(): Unit = {
+ assert(sc != null)
+ val dagOutputTracker = mapOutputTracker
+ val taskSchedulerImpl = new TaskSchedulerImpl(sc) {
+ override def submitTasks(taskSet: TaskSet) = {
+ super.submitTasks(taskSet)
+ taskSet.tasks.foreach(_.epoch = dagOutputTracker.getEpoch)
+ taskSets += taskSet
+ }
+
+ override def cancelTasks(stageId: Int, interruptThread: Boolean) {
+ cancelledStages += stageId
+ }
+ }
+ taskSchedulerImpl.initialize(new FakeSchedulerBackend)
+ scheduler = new DAGScheduler(
+ sc,
+ taskSchedulerImpl,
+ sc.listenerBus,
+ mapOutputTracker,
+ blockManagerMaster,
+ sc.env)
+ dagEventProcessLoopTester = new DAGSchedulerEventProcessLoopTester(scheduler)
+ }
+
+ test("Test dagScheduler.shouldUnregisterMapOutput with map stage not running") {
+ reInit()
+ val shuffleMapRdd = new MyRDD(sc, 2, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostB", reduceRdd.partitions.length))))
+ // The MapOutputTracker should know about both map output locations.
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId, 0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB"))
+
+ // The first result task fails, with a fetch failure for the output from the first mapper.
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0, 0, 0, "ignored"),
+ null))
+ assert(sparkListener.failedStages.contains(1))
+
+ val mapStatuses = mapOutputTracker.shuffleStatuses(shuffleId).mapStatuses
+ // unregisterMapOutput with a fetchFailed.
+ assert(mapStatuses.count(_ != null) === 1)
+ assert(mapStatuses(1).location === makeBlockManagerId("hostB"))
+ }
+
+ test("Test dagScheduler.shouldUnregisterMapOutput with map stage re-running") {
+ reInit()
+ val shuffleMapRdd = new MyRDD(sc, 3, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostB", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostC", reduceRdd.partitions.length))))
+ // The MapOutputTracker should know about both map output locations.
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId, 0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB", "hostC"))
+
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0, 0, 0, "ignored"),
+ null))
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(1),
+ FetchFailed(makeBlockManagerId("hostB"), shuffleId, 1, 1, 0, "ignored"),
+ null))
+
+ assert(sparkListener.failedStages.contains(1))
+
+ // Wait for a long time to make sure the map stage was resubmitted.
+ eventually(timeout(1000 milliseconds), interval(10 milliseconds)) {
+ assert(scheduler.runningStages.nonEmpty && taskSets.size == 3)
+ }
+
+ runEvent(makeCompletionEvent(
+ taskSets(2).tasks(0),
+ Success,
+ makeMapStatus("hostA", reduceRdd.partitions.length)))
+
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(1),
Review comment:
Oh, I'm sorry,i have update my ut, thanks, PTAL.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on a change in pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on a change in pull request #30716:
URL: https://github.com/apache/spark/pull/30716#discussion_r546631296
##########
File path: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
##########
@@ -2035,6 +2040,107 @@ class DAGSchedulerSuite extends SparkFunSuite with TempLocalSparkContext with Ti
assert(scheduler.activeJobs.isEmpty)
}
+ def reInit(): Unit = {
+ assert(sc != null)
+ val dagOutputTracker = mapOutputTracker
+ val taskSchedulerImpl = new TaskSchedulerImpl(sc) {
+ override def submitTasks(taskSet: TaskSet) = {
+ super.submitTasks(taskSet)
+ taskSet.tasks.foreach(_.epoch = dagOutputTracker.getEpoch)
+ taskSets += taskSet
+ }
+
+ override def cancelTasks(stageId: Int, interruptThread: Boolean) {
+ cancelledStages += stageId
+ }
+ }
+ taskSchedulerImpl.initialize(new FakeSchedulerBackend)
+ scheduler = new DAGScheduler(
+ sc,
+ taskSchedulerImpl,
+ sc.listenerBus,
+ mapOutputTracker,
+ blockManagerMaster,
+ sc.env)
+ dagEventProcessLoopTester = new DAGSchedulerEventProcessLoopTester(scheduler)
+ }
+
+ test("Test dagScheduler.shouldUnregisterMapOutput with map stage not running") {
+ reInit()
+ val shuffleMapRdd = new MyRDD(sc, 2, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostB", reduceRdd.partitions.length))))
+ // The MapOutputTracker should know about both map output locations.
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId, 0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB"))
+
+ // The first result task fails, with a fetch failure for the output from the first mapper.
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0, 0, 0, "ignored"),
+ null))
+ assert(sparkListener.failedStages.contains(1))
+
+ val mapStatuses = mapOutputTracker.shuffleStatuses(shuffleId).mapStatuses
+ // unregisterMapOutput with a fetchFailed.
+ assert(mapStatuses.count(_ != null) === 1)
+ assert(mapStatuses(1).location === makeBlockManagerId("hostB"))
+ }
+
+ test("Test dagScheduler.shouldUnregisterMapOutput with map stage re-running") {
+ reInit()
+ val shuffleMapRdd = new MyRDD(sc, 3, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostB", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostC", reduceRdd.partitions.length))))
+ // The MapOutputTracker should know about both map output locations.
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId, 0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB", "hostC"))
+
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0, 0, 0, "ignored"),
+ null))
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(1),
+ FetchFailed(makeBlockManagerId("hostB"), shuffleId, 1, 1, 0, "ignored"),
+ null))
+
+ assert(sparkListener.failedStages.contains(1))
+
+ // Wait for a long time to make sure the map stage was resubmitted.
+ eventually(timeout(1000 milliseconds), interval(10 milliseconds)) {
+ assert(scheduler.runningStages.nonEmpty && taskSets.size == 3)
+ }
+
+ runEvent(makeCompletionEvent(
+ taskSets(2).tasks(0),
+ Success,
+ makeMapStatus("hostA", reduceRdd.partitions.length)))
+
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(1),
Review comment:
Oh, I'm sorry,i have update my ut, thanks.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-743935128
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] github-actions[bot] closed pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #30716:
URL: https://github.com/apache/spark/pull/30716
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-746011514
> After taking another look, I think this problem can prevent by `shuffleFileLostEpoch`? All tasks from the failed reduce stage share the same epoch. So only the first fetch failed task can update the `shuffleFileLostEpoch` and unregister the map output.
This problem can not prevent by ` shuffleFileLostEpoch`, please take a look my uts, thanks.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 removed a comment on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 removed a comment on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-745828531
I've read the latest description. So the issue should only happen when speculation is enabled?
BTW, this PR looks more like an improvement to me instead of a bug fix.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-745828531
I've read the latest description. So the issue should only happen when speculation is enabled?
BTW, this PR looks more like an improvement to me instead of a bug fix.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-750233059
> After taking a look at the updated UT, I think the problem mentioned by @weixiuli is:
>
> 1. task X of Stage B encountered FetchFailure while fetching the shuffle data of Stage A at executor-0; Stage A unregisters the MapOutput for the mapIndex-0.
> 2. Stage A and Stage B both marked as failed and Stage A starts to rerun after a while.
> 3. task Y (run at executor-0 too) of rerun Stage A success and register its MapOutput for the same mapIndex-0
> 4. task Z of Stage B encountered FetchFailure at executor-0 too and Stage A unregister the MapOutput of mapIndex-1 again
>
> I think what @weixiuli is trying to do here is to avoid unregistering the MapOutput in step 4 again.
>
> However, I wonder whether it could happen in a real job. Or say, it's really a corner case. I think tasks like task Z usually failed early before the rerun task Y success.
This is a potential problem, especially if the network is poor and resources are scarce.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-748901842
> I am a bit unclear on this - trying to understand what is the race condition here.
> Particularly given [this issue in pr](https://github.com/apache/spark/pull/30716/files#diff-85de35b2e85646ed499c545a3be1cd3ffd525a88aae835a9c621f877eebadcb6R267), I am not sure what the actual behavior being tested here is.
>
> Thoughts @Ngone51, @weixiuli ?
@mridulm Please take a look the PR description , thanks.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-743498865
> Doesn't the check below prevents the case you mentioned in the PR description?
>
> https://github.com/apache/spark/blob/485145326a9c97ede260b0e267ee116f182cfd56/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1706-L1709
Thanks for your review, this PR avoid calling unregisterMapOutput when` failedStage.latestInfo.attemptNumber == task.stageAttemptId` and the new mapstage has been rerunning.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-745849444
After taking another look, I think this problem can prevent by `shuffleFileLostEpoch`? All tasks from the failed reduce stage share the same epoch. So only the first fetch failed task can update the `shuffleFileLostEpoch` and unregister the map output.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-745721466
> > this PR avoid calling unregisterMapOutput when failedStage.latestInfo.attemptNumber == task.stageAttemptId and the new mapstage has been rerunning.
>
> Could you explain more about the race condition? I still can't get what's going on here.
@Ngone51 I have updated the PR description, PTAL.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-743956322
retest this please
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on a change in pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on a change in pull request #30716:
URL: https://github.com/apache/spark/pull/30716#discussion_r557333541
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -251,6 +251,26 @@ private[spark] class DAGScheduler(
private val pushBasedShuffleEnabled = Utils.isPushBasedShuffleEnabled(sc.getConf)
+ private def shouldUnregisterMapOutput(mapStage: Stage, fetchFailed: FetchFailed,
+ stageId: Int, stageAttemptId: Int): Boolean = {
+ if (!runningStages.contains(mapStage)) {
+ true
+ } else if (taskScheduler == null) {
+ true
+ } else {
+ val latestInfo = mapStage.latestInfo
+ if (latestInfo == null) {
+ true
+ } else {
+ taskScheduler.taskSetManagerForAttempt(
+ latestInfo.stageId, latestInfo.attemptNumber()) match {
+ case Some(tsm: TaskSetManager) => !tsm.hasPartitionId(fetchFailed.mapId.toInt)
+ case _ => true
Review comment:
Compared with before, I think this is acceptable, do you have any idea?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-749991166
After taking a look at the updated UT, I think the problem mentioned by @weixiuli is:
1) task X of Stage B encountered FetchFailure while fetching the shuffle data of Stage A at executor-0; Stage A unregisters the MapOutput for the mapIndex-0.
2) Stage A and Stage B both marked as failed and Stage A starts to rerun after a while.
3) task Y (run at executor-0 too) of rerun Stage A success and register its MapOutput for the same mapIndex-0
4) task Z of Stage B encountered FetchFailure at executor-0 too and Stage A unregister the MapOutput of mapIndex-1 again
I think what @weixiuli is trying to do here is to avoid unregistering the MapOutput in step 4 again.
However, I wonder whether it could happen in a real job. Or say, it's really a corner case. I think tasks like task Z usually failed early before the rerun task Y success.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #30716:
URL: https://github.com/apache/spark/pull/30716#discussion_r549996831
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -251,6 +251,26 @@ private[spark] class DAGScheduler(
private val pushBasedShuffleEnabled = Utils.isPushBasedShuffleEnabled(sc.getConf)
+ private def shouldUnregisterMapOutput(mapStage: Stage, fetchFailed: FetchFailed,
+ stageId: Int, stageAttemptId: Int): Boolean = {
+ if (!runningStages.contains(mapStage)) {
+ true
+ } else if (taskScheduler == null) {
+ true
+ } else {
+ val latestInfo = mapStage.latestInfo
+ if (latestInfo == null) {
+ true
+ } else {
+ taskScheduler.taskSetManagerForAttempt(
+ latestInfo.stageId, latestInfo.attemptNumber()) match {
+ case Some(tsm: TaskSetManager) => !tsm.hasPartitionId(fetchFailed.mapId.toInt)
+ case _ => true
Review comment:
In the corner case, e.g., the rerun stage may only need to rerun one task and the task finished before this FetchFailure...which means the rerun stage attempt could be removed from `taskSetManagerForAttempt`...then, I don't think we should return `true` here..
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -251,6 +251,26 @@ private[spark] class DAGScheduler(
private val pushBasedShuffleEnabled = Utils.isPushBasedShuffleEnabled(sc.getConf)
+ private def shouldUnregisterMapOutput(mapStage: Stage, fetchFailed: FetchFailed,
+ stageId: Int, stageAttemptId: Int): Boolean = {
+ if (!runningStages.contains(mapStage)) {
+ true
+ } else if (taskScheduler == null) {
+ true
+ } else {
+ val latestInfo = mapStage.latestInfo
+ if (latestInfo == null) {
+ true
+ } else {
+ taskScheduler.taskSetManagerForAttempt(
+ latestInfo.stageId, latestInfo.attemptNumber()) match {
+ case Some(tsm: TaskSetManager) => !tsm.hasPartitionId(fetchFailed.mapId.toInt)
Review comment:
I think you actually want `fetchFailed.mapIndex` rather than `fetchFailed.mapId`?
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -1731,8 +1751,14 @@ private[spark] class DAGScheduler(
// resubmitted stage attempt.
mapOutputTracker.unregisterAllMapOutput(shuffleId)
} else if (mapIndex != -1) {
- // Mark the map whose fetch failed as broken in the map stage
- mapOutputTracker.unregisterMapOutput(shuffleId, mapIndex, bmAddress)
Review comment:
Probably, comparing the `mapId` of the MapStatus should be enough for this case. But it should only work when we use the new shuffle fetch protocol (enabled by default). For example,
```scala
val mapStatus = mapOutputTracker.getMapOutput(shuffleId, mapIndex)
if (mapStatus.mapId == fetchFailed.mapId) {
mapOutputTracker.unregisterMapOutput(shuffleId, mapIndex, bmAddress)
}
```
For the old protocol, I think we can at least compare `bmAddress` to prevent most cases. Although, it still cannot fix the case where `bmAddress` is the same.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm edited a comment on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
mridulm edited a comment on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-747839969
I am a bit unclear on this - trying to understand what is the race condition here.
Particularly given [this issue in pr](https://github.com/apache/spark/pull/30716/files#diff-85de35b2e85646ed499c545a3be1cd3ffd525a88aae835a9c621f877eebadcb6R267), I am not sure what the actual behavior being tested here is.
Thoughts @Ngone51, @weixiuli ?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #30716:
URL: https://github.com/apache/spark/pull/30716#discussion_r544765448
##########
File path: core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
##########
@@ -2035,6 +2040,107 @@ class DAGSchedulerSuite extends SparkFunSuite with TempLocalSparkContext with Ti
assert(scheduler.activeJobs.isEmpty)
}
+ def reInit(): Unit = {
+ assert(sc != null)
+ val dagOutputTracker = mapOutputTracker
+ val taskSchedulerImpl = new TaskSchedulerImpl(sc) {
+ override def submitTasks(taskSet: TaskSet) = {
+ super.submitTasks(taskSet)
+ taskSet.tasks.foreach(_.epoch = dagOutputTracker.getEpoch)
+ taskSets += taskSet
+ }
+
+ override def cancelTasks(stageId: Int, interruptThread: Boolean) {
+ cancelledStages += stageId
+ }
+ }
+ taskSchedulerImpl.initialize(new FakeSchedulerBackend)
+ scheduler = new DAGScheduler(
+ sc,
+ taskSchedulerImpl,
+ sc.listenerBus,
+ mapOutputTracker,
+ blockManagerMaster,
+ sc.env)
+ dagEventProcessLoopTester = new DAGSchedulerEventProcessLoopTester(scheduler)
+ }
+
+ test("Test dagScheduler.shouldUnregisterMapOutput with map stage not running") {
+ reInit()
+ val shuffleMapRdd = new MyRDD(sc, 2, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostB", reduceRdd.partitions.length))))
+ // The MapOutputTracker should know about both map output locations.
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId, 0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB"))
+
+ // The first result task fails, with a fetch failure for the output from the first mapper.
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0, 0, 0, "ignored"),
+ null))
+ assert(sparkListener.failedStages.contains(1))
+
+ val mapStatuses = mapOutputTracker.shuffleStatuses(shuffleId).mapStatuses
+ // unregisterMapOutput with a fetchFailed.
+ assert(mapStatuses.count(_ != null) === 1)
+ assert(mapStatuses(1).location === makeBlockManagerId("hostB"))
+ }
+
+ test("Test dagScheduler.shouldUnregisterMapOutput with map stage re-running") {
+ reInit()
+ val shuffleMapRdd = new MyRDD(sc, 3, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val shuffleId = shuffleDep.shuffleId
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ submit(reduceRdd, Array(0, 1))
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostB", reduceRdd.partitions.length)),
+ (Success, makeMapStatus("hostC", reduceRdd.partitions.length))))
+ // The MapOutputTracker should know about both map output locations.
+ assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId, 0).map(_._1.host).toSet ===
+ HashSet("hostA", "hostB", "hostC"))
+
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(0),
+ FetchFailed(makeBlockManagerId("hostA"), shuffleId, 0, 0, 0, "ignored"),
+ null))
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(1),
+ FetchFailed(makeBlockManagerId("hostB"), shuffleId, 1, 1, 0, "ignored"),
+ null))
+
+ assert(sparkListener.failedStages.contains(1))
+
+ // Wait for a long time to make sure the map stage was resubmitted.
+ eventually(timeout(1000 milliseconds), interval(10 milliseconds)) {
+ assert(scheduler.runningStages.nonEmpty && taskSets.size == 3)
+ }
+
+ runEvent(makeCompletionEvent(
+ taskSets(2).tasks(0),
+ Success,
+ makeMapStatus("hostA", reduceRdd.partitions.length)))
+
+ runEvent(makeCompletionEvent(
+ taskSets(1).tasks(1),
Review comment:
`taskSets(1).tasks(1)` already completed with `FetchFailure` above, why it can complete 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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-751131070
@Ngone51 @mridulm @jiangxb1987 @dongjoon-hyun PTAL.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-742911654
Can one of the admins verify this patch?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] github-actions[bot] commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-830938117
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli edited a comment on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli edited a comment on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-743935047
@Ngone51 PTAL, ping @dongjoon-hyun Kindly review.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-745829190
cc @tgravescs @mridulm @jiangxb1987
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-753646067
Can one of the admins verify this patch?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-742911654
Can one of the admins verify this patch?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] weixiuli commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
weixiuli commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-749343263
@Ngone51 @mridulm @jiangxb1987 @dongjoon-hyun PTAL.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-744211042
> this PR avoid calling unregisterMapOutput when failedStage.latestInfo.attemptNumber == task.stageAttemptId and the new mapstage has been rerunning.
Could you explain more about the race condition? I still can't get what's going on 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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #30716: [SPARK-33747][CORE] Avoid calling unregisterMapOutput when the map stage is being rerunning.
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #30716:
URL: https://github.com/apache/spark/pull/30716#issuecomment-747165924
Besides, I tested `Test dagScheduler.shouldUnregisterMapOutput with map stage re-running` against `Master` branch. It passed without any failure.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org