[
https://issues.apache.org/jira/browse/SPARK-58883?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yang Jie updated SPARK-58883:
-----------------------------
Description:
Delete the *last* partition file from a checkpoint directory and reading it
back silently produces an RDD with fewer partitions, losing that partition's
rows with no error.
{code:none}
sc.setCheckpointDir("/data/cp")
val rdd = sc.makeRDD(1 to 20, numSlices = 4)
rdd.checkpoint()
rdd.collect() // 20 rows,
part-00000..part-00003 written
// part-00003 is lost: pruned by a cleanup job, an incomplete copy, a partial
upload
sc.checkpointFile[Int]("/data/cp/rdd-3").count() // 15, no error
{code}
{{ReliableCheckpointRDD.getPartitions}} validates only that the {{part-*}}
names it finds are contiguous from {{part-00000}}. Dropping a file from the
middle breaks contiguity and raises {{INVALID_CHECKPOINT_DIRECTORY}}, but
dropping the trailing one leaves the rest contiguous, so the loop passes and
{{Array.tabulate(inputFiles.length)}} yields one partition per surviving file.
On the write path this is caught: {{writeRDDToCheckpointDirectory}} compares
{{newRDD.partitions.length}} against {{originalRDD.partitions.length}} and
raises {{CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH}}. On the read-back path
through {{SparkContext.checkpointFile}} there is no original RDD to compare
against. That path is how Spark Streaming rebuilds {{generatedRDDs}} during
recovery ({{DStreamCheckpointData.restore}}), so a truncated checkpoint means
silently incomplete recovered state.
{{getPartitions}}' own scaladoc states the gap: "Since the original RDD may
belong to a prior application, there is no way to know a priori the number of
partitions to expect."
Detecting this requires persisting the expected partition count at write time
and comparing on read, i.e. a new metadata file in the checkpoint directory.
Design points that need deciding:
* the file name, and whether a failure to write it is fatal
* backward compatibility: directories written by earlier versions have no such
file, so a missing file must be tolerated. {{_partitioner}} is the precedent
({{writePartitionerToCheckpointDir}} / {{readCheckpointedPartitionerFile}},
which returns {{None}} on {{FileNotFoundException}})
* which error condition a mismatch raises on the read path.
{{CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH}} currently means "the directory just
written disagrees with the RDD that produced it", which does not quite fit a
read of someone else's directory
* whether an empty directory (no {{part-*}} at all, currently a 0-partition
RDD) should be treated the same way
Found while reviewing SPARK-58770 (https://github.com/apache/spark/pull/58004).
Deliberately left out of that PR, which only assigns names to existing error
conditions: this needs a new on-disk format and its own compatibility
discussion. That PR's message for {{INVALID_CHECKPOINT_DIRECTORY}} was trimmed
to claim only the contiguity the loop actually checks, so it no longer implies
the whole directory is validated.
> Reading a truncated checkpoint directory silently yields fewer partitions
> -------------------------------------------------------------------------
>
> Key: SPARK-58883
> URL: https://issues.apache.org/jira/browse/SPARK-58883
> Project: Spark
> Issue Type: Bug
> Components: Spark Core
> Affects Versions: 5.0.0
> Reporter: Yang Jie
> Priority: Major
>
> Delete the *last* partition file from a checkpoint directory and reading it
> back silently produces an RDD with fewer partitions, losing that partition's
> rows with no error.
> {code:none}
> sc.setCheckpointDir("/data/cp")
> val rdd = sc.makeRDD(1 to 20, numSlices = 4)
> rdd.checkpoint()
> rdd.collect() // 20 rows,
> part-00000..part-00003 written
> // part-00003 is lost: pruned by a cleanup job, an incomplete copy, a partial
> upload
> sc.checkpointFile[Int]("/data/cp/rdd-3").count() // 15, no error
> {code}
> {{ReliableCheckpointRDD.getPartitions}} validates only that the {{part-*}}
> names it finds are contiguous from {{part-00000}}. Dropping a file from the
> middle breaks contiguity and raises {{INVALID_CHECKPOINT_DIRECTORY}}, but
> dropping the trailing one leaves the rest contiguous, so the loop passes and
> {{Array.tabulate(inputFiles.length)}} yields one partition per surviving file.
> On the write path this is caught: {{writeRDDToCheckpointDirectory}} compares
> {{newRDD.partitions.length}} against {{originalRDD.partitions.length}} and
> raises {{CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH}}. On the read-back path
> through {{SparkContext.checkpointFile}} there is no original RDD to compare
> against. That path is how Spark Streaming rebuilds {{generatedRDDs}} during
> recovery ({{DStreamCheckpointData.restore}}), so a truncated checkpoint means
> silently incomplete recovered state.
> {{getPartitions}}' own scaladoc states the gap: "Since the original RDD may
> belong to a prior application, there is no way to know a priori the number of
> partitions to expect."
> Detecting this requires persisting the expected partition count at write time
> and comparing on read, i.e. a new metadata file in the checkpoint directory.
> Design points that need deciding:
> * the file name, and whether a failure to write it is fatal
> * backward compatibility: directories written by earlier versions have no
> such file, so a missing file must be tolerated. {{_partitioner}} is the
> precedent ({{writePartitionerToCheckpointDir}} /
> {{readCheckpointedPartitionerFile}}, which returns {{None}} on
> {{FileNotFoundException}})
> * which error condition a mismatch raises on the read path.
> {{CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH}} currently means "the directory
> just written disagrees with the RDD that produced it", which does not quite
> fit a read of someone else's directory
> * whether an empty directory (no {{part-*}} at all, currently a 0-partition
> RDD) should be treated the same way
> Found while reviewing SPARK-58770
> (https://github.com/apache/spark/pull/58004). Deliberately left out of that
> PR, which only assigns names to existing error conditions: this needs a new
> on-disk format and its own compatibility discussion. That PR's message for
> {{INVALID_CHECKPOINT_DIRECTORY}} was trimmed to claim only the contiguity the
> loop actually checks, so it no longer implies the whole directory is
> validated.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
