James Willis created SPARK-58750:
------------------------------------
Summary: 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: 4.0.3, 4.1.2, 4.1.1, 4.2.0, 3.5.8, 4.0.2, 3.5.7, 4.0.1,
4.0.0, 3.5.6, 4.1.0, 3.5.5, 3.5.4, 3.5.3, 3.4.4, 3.4.3, 3.5.2, 3.3.4, 3.5.1,
3.5.0, 3.4.1, 3.4.0, 3.3.2, 3.4.2, 3.3.3, 3.2.4, 3.2.3, 3.3.1, 3.2.2, 3.3.0,
3.2.1, 3.2.0, 3.5.9, 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
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]
