[
https://issues.apache.org/jira/browse/SPARK-58750?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58750:
-----------------------------------
Labels: pull-request-available (was: )
> RDD checkpoint to S3A fails unrecoverably under speculation/task retry
> -----------------------------------------------------------------------
>
> Key: SPARK-58750
> URL: https://issues.apache.org/jira/browse/SPARK-58750
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 3.2.0, 3.2.1, 3.3.0, 3.2.2, 3.3.1, 3.2.3, 3.2.4, 3.3.3,
> 3.4.2, 3.3.2, 3.4.0, 3.4.1, 3.5.0, 3.5.1, 3.3.4, 3.5.2, 3.4.3, 3.4.4, 3.5.3,
> 3.5.4, 3.5.5, 4.1.0, 3.5.6, 4.0.0, 4.0.1, 3.5.7, 4.0.2, 3.5.8, 4.2.0, 4.1.1,
> 4.1.2, 3.5.9, 4.0.3, 4.2.1, 3.5.10
> Environment: I produced this error in Spark 4.0.1 on Kubernetes,
> Hadoop 3.4.1 (S3A), with {{spark.speculation=true}}
> Reporter: James Willis
> Priority: Major
> Labels: pull-request-available
>
> h2. Summary
> {{ReliableCheckpointRDD.writePartitionToCheckpointFile}} assumes HDFS rename
> semantics: it handles {{fs.rename()}} returning {{false}} when the
> destination exists, but does not handle the rename {*}throwing{*}. Since
> HADOOP-16721 (Hadoop 3.3.1), S3A deliberately raises
> {{FileAlreadyExistsException}} when the rename destination is an existing
> file, instead of returning {{{}false{}}}. ABFS behaves the same way.
> Under speculative execution, two attempts of the same checkpoint task race to
> rename their attempt-temp file onto the same final part file. On HDFS the
> loser gets {{rename() == false}} and Spark correctly treats it as "some other
> copy of this task must've finished before us". On S3A the loser gets an
> unhandled {{{}FileAlreadyExistsException{}}}, which fails the task even
> though the data is already written. Because the destination file now
> permanently exists, {*}every subsequent retry of that task fails on the same
> rename{*}, so {{spark.task.maxFailures}} is always exhausted and the job
> aborts.
> Every Spark release since the default Hadoop dependency moved to 3.3.1+
> (Spark 3.2.0) is affected when checkpointing to S3A or ABFS with speculation
> enabled. The code is unchanged in master.
> h2. Observed failure (production, Spark 4.0.1 / Hadoop 3.4.1)
> {noformat}
> org.apache.hadoop.fs.FileAlreadyExistsException: Failed to rename
>
> s3://<bucket>/<prefix>/spark-checkpoints/44151dc0-c680-446d-8556-f6aef47d73dd/rdd-207/.part-00379-attempt-25364
> to
> s3://<bucket>/<prefix>/spark-checkpoints/44151dc0-c680-446d-8556-f6aef47d73dd/rdd-207/part-00379;
> destination file exists
> at
> org.apache.hadoop.fs.s3a.S3AFileSystem.initiateRename(S3AFileSystem.java:2468)
> at
> org.apache.hadoop.fs.s3a.S3AFileSystem.innerRename(S3AFileSystem.java:2533)
> at
> org.apache.hadoop.fs.s3a.S3AFileSystem.lambda$rename$6(S3AFileSystem.java:2394)
> ...
> at
> org.apache.hadoop.fs.s3a.S3AFileSystem.rename(S3AFileSystem.java:2392)
> at
> org.apache.spark.rdd.ReliableCheckpointRDD$.writePartitionToCheckpointFile(ReliableCheckpointRDD.scala:229)
> at
> org.apache.spark.rdd.ReliableCheckpointRDD$.$anonfun$writeRDDToCheckpointDirectory$1(ReliableCheckpointRDD.scala:168)
> at org.apache.spark.scheduler.Task.run(Task.scala:147)
> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:650)
> {noformat}
> followed by:
> {noformat}
> ERROR TaskSetManager: Task 379 in stage 105.0 failed 4 times; aborting job
> {noformat}
> Speculative kills were visible throughout the run; task 379.0 was still
> running when attempt 379.4 hit the already-committed part file.
> h2. Root cause
> {{writePartitionToCheckpointFile}} (master):
> {code:scala}
> if (!fs.rename(tempOutputPath, finalOutputPath)) {
> if (!fs.exists(finalOutputPath)) {
> logInfo(...)
> fs.delete(tempOutputPath, false)
> throw SparkCoreErrors.checkpointFailedToSaveError(ctx.attemptNumber(),
> finalOutputPath)
> } else {
> // Some other copy of this task must've finished before us and renamed
> it
> ...
> }
> }
> {code}
> The dest-exists race is only handled when {{rename()}} reports it by
> returning {{{}false{}}}. The Hadoop FileSystem specification does not
> guarantee that; HADOOP-16721 explicitly changed S3A "away from consistency
> with HDFS to one where applications find out what is wrong", raising
> {{FileAlreadyExistsException}} for a file destination. Structured Streaming's
> {{CheckpointFileManager}} was hardened for divergent rename semantics (e.g.
> SPARK-40826); the RDD checkpoint writer was not.
> h2. Reproduction
> # Configure an S3A (or ABFS) checkpoint dir, Hadoop >= 3.3.1,
> {{spark.speculation=true}} (aggressive settings help:
> {{{}spark.speculation.multiplier=1.0{}}},
> {{{}spark.speculation.quantile=0.1{}}})
> # {{sc.setCheckpointDir("s3a://bucket/checkpoints")}}
> # Checkpoint an RDD with skewed/slow partitions so speculative attempts
> launch, e.g. a mapPartitions that sleeps on a subset of partitions; call
> {{rdd.checkpoint()}} then an action
> # A speculative attempt losing the rename race fails the task; all retries
> of that task then fail deterministically and the job aborts
> Also reproducible without speculation when a task fails *after* its rename
> commits and the task is retried (e.g. executor lost during post-write
> bookkeeping): the retry finds the destination and dies the same way.
> h2. Proposed fix
> Fold the exception into the existing false-return path, which already
> re-checks the destination defensively:
> {code:scala}
> val renamed = try {
> fs.rename(tempOutputPath, finalOutputPath)
> } catch {
> case _: FileAlreadyExistsException => false
> }
> if (!renamed) {
> // existing logic: if finalOutputPath exists, another attempt committed —
> clean up temp and
> // succeed; otherwise delete temp and throw checkpointFailedToSaveError
> }
> {code}
> Semantics are unchanged relative to HDFS: whichever attempt commits the part
> file wins, exactly as the winner-takes-all rename behaves today on
> filesystems that return {{{}false{}}}. I can submit a PR with this change and
> a test.
> h2. Workaround
> None good: speculation cannot be disabled per-job, so users must disable it
> application-wide, checkpoint to HDFS-semantics storage, or replace
> {{checkpoint()}} with a manual write/read-back through a file-source write
> (which goes through the output commit coordinator and is speculation-safe).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
