Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SPARK-28699][Core] Cache an indeterminate RDD could lead to incorrect result while stage rerun #25420

Closed
wants to merge 2 commits into from

Conversation

xuanyuanking
Copy link
Member

@xuanyuanking xuanyuanking commented Aug 12, 2019

What changes were proposed in this pull request?

It's another case for the indeterminate stage/RDD rerun while stage rerun happened. In the CachedRDDBuilder, we miss tracking the isOrderSensitive characteristic to the newly created MapPartitionsRDD.
This patch just a safeguard, if we need the support for stage rerunning, it should be done after #24892.

How was this patch tested?

Integrated test with an exception, instead of the wrong answer.

import scala.sys.process._
import org.apache.spark.TaskContext

val res = spark.range(0, 10000 * 10000, 1).map{ x => (x % 1000, x)}
// kill an executor in the stage that performs repartition(239)
val df = res.repartition(113).cache.repartition(239).map { x =>
  if (TaskContext.get.attemptNumber == 0 && TaskContext.get.partitionId < 1 && TaskContext.get.stageAttemptNumber == 0) {
    throw new Exception("pkill -f -n java".!!)
  }
  x
}
val r2 = df.distinct.count()

(cherry picked from commit b543fab)
Signed-off-by: Yuanjian Li <xyliyuanjian@gmail.com>
@@ -2710,7 +2710,7 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with TimeLi
assert(countSubmittedMapStageAttempts() === 2)
}

test("SPARK-23207: retry all the succeeding stages when the map stage is indeterminate") {
ignore("SPARK-23207: retry all the succeeding stages when the map stage is indeterminate") {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ignore this for the behavior change, as the approach now, we need to abort the stage of the current mapStage.
As we will finally support stage rerun, I suggest to skip this behavior, we can directly support the cache scenario after SPARK-25341 merged.

@@ -131,7 +131,7 @@ case class CachedRDDBuilder(

def hasNext: Boolean = rowIterator.hasNext
}
}.persist(storageLevel)
}, isOrderSensitive = true).persist(storageLevel)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't quite get it. The isOrderSensitive flag is used to describe the map function. Why the map function is order sensitive?

@SparkQA
Copy link

SparkQA commented Aug 12, 2019

Test build #108978 has finished for PR 25420 at commit b3ea90f.

  • This patch fails Spark unit tests.
  • This patch merges cleanly.
  • This patch adds no public classes.

@@ -86,7 +86,7 @@ case class CachedRDDBuilder(

private def buildBuffers(): RDD[CachedBatch] = {
val output = cachedPlan.output
val cached = cachedPlan.execute().mapPartitionsInternal { rowIterator =>
val cached = cachedPlan.execute().mapPartitionsWithIndexInternal({ (_, rowIterator) =>
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we only run cached plan once. I think it should be determinate? If so, this map should not be order sensitive effectively?

@dongjoon-hyun
Copy link
Member

Retest this please.

@dongjoon-hyun
Copy link
Member

@xuanyuanking Can we have a test case to prevent a future regression on this?

@SparkQA
Copy link

SparkQA commented Aug 12, 2019

Test build #108993 has finished for PR 25420 at commit b3ea90f.

  • This patch passes all tests.
  • This patch merges cleanly.
  • This patch adds no public classes.

@xuanyuanking
Copy link
Member Author

xuanyuanking commented Aug 19, 2019

Sorry for the late. Thanks for the comment and review! After taking a further investigation, I found the root cause for this issue is about radix sort in ShuffleExchangeExec, please review it in #25491.
The bug fix of DAGScheduler is still needed, I'll give another follow-up PR and add UT for it. Done in #25498.
After all the fix has done, I'll give detailed explanation in the JIRA description.

@xuanyuanking xuanyuanking deleted the SPARK-28699 branch August 19, 2019 09:05
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
5 participants