How can I reproduce the indeterminacy exception in Spark?

Viewed 16

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?

0 Answers
Related