Here's the indeterminacy I'm looking to reproduce: ERROR org.apache.spark.deploy.yarn.Client: Application diagnostics message: User class threw exception: org.apache.spark.SparkException: Job aborted due to stage failure: A shuffle map stage with indeterminate output was failed and retried. However, Spark cannot rollback the ShuffleMapStage 401 to re-process the input data, and has to fail this job. Please eliminate the indeterminacy by checkpointing the RDD before repartition and try again.
As I understand, this happens because a part of logic is not deterministic (e.g. using LocalTime.now() or using scala.util.Random). I wrote a piece of code that was retried and confirmed to be non-deterministic, but it still didn't fail with the indeterminacy exception. In fact, it succeeded.
import spark.implicits._
val data = spark
.range(0, 100, 1)
.map(identity)
.repartition(7)
.map { x =>
if ((x % 10) == 0) {
Thread.sleep(1000)
}
val result = Entry("key" + x, x + RichDate.now.timestamp)
println(x + " " + result)
result
}
.map { item =>
if (TaskContext.get.attemptNumber == 0 && TaskContext.get.partitionId > 0 && TaskContext.get
.partitionId() < 3) {
println(
"Throw " + TaskContext.get.attemptNumber() + " " + TaskContext.get
.partitionId() + " " + TaskContext.get.stageAttemptNumber()
)
throw new Exception("pkill -f -n java".!!)
}
Output(item.toString)
}
// write to S3
println(x + " " + result) confirms that the specific part of code is retried and produces different result at the second try.
What did I do wrong? And does anyone have a sample code that can reproduce the indeterminacy exception?