LuciferYang commented on code in PR #58167:
URL: https://github.com/apache/spark/pull/58167#discussion_r3860940006
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -83,6 +85,16 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
throw SparkCoreErrors.invalidCheckpointDirectoryError(path,
expectedFileName)
}
}
+ // If a partition-count metadata file is present, verify no trailing files
are missing.
+ // Directories written by earlier Spark versions have no such file; a
missing file is
+ // silently tolerated for backward compatibility. See SPARK-58883.
+ ReliableCheckpointRDD.readPartitionCountFromCheckpointDir(context,
checkpointPath)
+ .foreach { expected =>
+ if (inputFiles.length != expected) {
Review Comment:
Writing and reading `_num_partitions` both tolerate any failure (`:312`,
`:346`), but acting on it (`:93`) is fatal and consults no config, and the diff
does not touch `internal/config`. The analogous local-checkpoint integrity
feature in this same release ships behind
`spark.checkpoint.local.verifyChecksum.enabled`, an `.internal()` flag
defaulting to true.
There is no API-level way around the throw either; neither version of
`checkpointFile` is public. What is left is deleting `_num_partitions` from
storage, and since the error names no directory, that means deleting it from
every `rdd-*`, which turns the check off wholesale. Until someone does that, a
streaming app fails on every restart. Gating this comparison the same way would
cover it.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -268,6 +285,70 @@ private[spark] object ReliableCheckpointRDD extends
Logging {
}
}
+ /**
+ * Write the partition count of the checkpointed RDD to the checkpoint
directory so that
+ * a later read via [[SparkContext.checkpointFile]] can detect a truncated
directory.
+ * This is done on a best-effort basis; any exception is caught, logged and
ignored so that
+ * an inability to write the file does not prevent checkpointing. See
SPARK-58883.
+ */
+ private def writePartitionCountToCheckpointDir(
+ sc: SparkContext, partitionCount: Int, checkpointDirPath: Path): Unit = {
+ try {
+ val countFilePath = new Path(checkpointDirPath,
checkpointPartitionCountFileName())
+ val bufferSize = sc.conf.get(BUFFER_SIZE)
+ val fs = countFilePath.getFileSystem(sc.hadoopConfiguration)
+ // overwrite = false: matches _partitioner's write helper; a second
checkpoint to the
+ // same directory would fail here (caught and logged below), which is
acceptable.
+ val fileOutputStream = fs.create(countFilePath, false, bufferSize)
+ val serializer = SparkEnv.get.serializer.newInstance()
+ val serializeStream = serializer.serializeStream(fileOutputStream)
+ Utils.tryWithSafeFinally {
+ serializeStream.writeObject(partitionCount)
+ } {
+ serializeStream.close()
+ }
+ logDebug(s"Written partition count $partitionCount to $countFilePath")
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Error writing partition count to ${MDC(PATH,
checkpointDirPath)}")
+ }
+ }
+
+ /**
+ * Read the expected partition count from the checkpoint directory metadata
file, if present.
+ * Returns [[None]] when the file is absent (checkpoint written by an older
Spark version)
+ * or unreadable, so callers must tolerate a missing value. See SPARK-58883.
+ */
+ private def readPartitionCountFromCheckpointDir(
+ sc: SparkContext, checkpointDirPath: String): Option[Int] = {
+ try {
+ val bufferSize = sc.conf.get(BUFFER_SIZE)
+ val countFilePath = new Path(checkpointDirPath,
checkpointPartitionCountFileName())
+ val fs = countFilePath.getFileSystem(sc.hadoopConfiguration)
+ val fileInputStream = fs.open(countFilePath, bufferSize)
+ val serializer = SparkEnv.get.serializer.newInstance()
+ val count = Utils.tryWithSafeFinally {
+ val deserializeStream = serializer.deserializeStream(fileInputStream)
+ Utils.tryWithSafeFinally {
+ deserializeStream.readObject[Int]()
+ } {
+ deserializeStream.close()
+ }
+ } {
+ fileInputStream.close()
+ }
+ logDebug(s"Read partition count $count from $countFilePath")
+ Some(count)
+ } catch {
+ case _: FileNotFoundException =>
+ logDebug(s"No partition count file in $checkpointDirPath (older
checkpoint)")
+ None
+ case NonFatal(e) =>
Review Comment:
`_num_partitions` is written with no atomic replace: part files go to
`.part-NNNNN-attempt-K` and are renamed (`:217`, `:246`), while this one, like
`_partitioner`, is a plain `fs.create` (`:302`, `:272`). The difference is the
consequence: losing the partitioner costs a shuffle, losing this file turns off
the check the PR just added. If the driver dies mid-write, or the storage
truncates it, the read at `:346` swallows the `NonFatal` into `None` and one
WARN is all that is left. That is the same storage this PR assumes can drop a
trailing file.
Routing the Int through `DataOutputStream.writeInt` plus a format version,
written to a temp path and renamed, makes it a fixed 8 bytes whose length shows
a torn write.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -268,6 +285,70 @@ private[spark] object ReliableCheckpointRDD extends
Logging {
}
}
+ /**
+ * Write the partition count of the checkpointed RDD to the checkpoint
directory so that
+ * a later read via [[SparkContext.checkpointFile]] can detect a truncated
directory.
+ * This is done on a best-effort basis; any exception is caught, logged and
ignored so that
+ * an inability to write the file does not prevent checkpointing. See
SPARK-58883.
+ */
+ private def writePartitionCountToCheckpointDir(
+ sc: SparkContext, partitionCount: Int, checkpointDirPath: Path): Unit = {
+ try {
+ val countFilePath = new Path(checkpointDirPath,
checkpointPartitionCountFileName())
+ val bufferSize = sc.conf.get(BUFFER_SIZE)
+ val fs = countFilePath.getFileSystem(sc.hadoopConfiguration)
+ // overwrite = false: matches _partitioner's write helper; a second
checkpoint to the
+ // same directory would fail here (caught and logged below), which is
acceptable.
+ val fileOutputStream = fs.create(countFilePath, false, bufferSize)
+ val serializer = SparkEnv.get.serializer.newInstance()
+ val serializeStream = serializer.serializeStream(fileOutputStream)
+ Utils.tryWithSafeFinally {
+ serializeStream.writeObject(partitionCount)
+ } {
+ serializeStream.close()
+ }
+ logDebug(s"Written partition count $partitionCount to $countFilePath")
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Error writing partition count to ${MDC(PATH,
checkpointDirPath)}")
Review Comment:
The catch in `writePartitionCountToCheckpointDir` never uses `e` (`:312`),
so a failed write leaves one WARN with no cause. The read side treats a missing
`_num_partitions` as a directory written by an older Spark (`:343`), so once
the file fails to appear, truncation detection stays off for that directory
while `rdd.checkpoint()` still returns success.
The minimum is to pass `e` to `logWarning`, add the part-file count found in
the directory, and say in the scaladoc that a failed write leaves the check
inactive. Letting the write throw is the stronger option, but it is not free:
`doCheckpoint()` runs after `dagScheduler.runJob` has returned, so throwing
fails an action whose result is already computed, and `cpState` stays at
`CheckpointingInProgress`.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -83,6 +85,16 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
throw SparkCoreErrors.invalidCheckpointDirectoryError(path,
expectedFileName)
}
}
+ // If a partition-count metadata file is present, verify no trailing files
are missing.
+ // Directories written by earlier Spark versions have no such file; a
missing file is
+ // silently tolerated for backward compatibility. See SPARK-58883.
+ ReliableCheckpointRDD.readPartitionCountFromCheckpointDir(context,
checkpointPath)
+ .foreach { expected =>
+ if (inputFiles.length != expected) {
+ throw
SparkCoreErrors.checkpointRDDHasDifferentNumberOfPartitionsFromOriginalRDDError(
+ id, expected, id, inputFiles.length)
Review Comment:
`:95` passes the same `id` into both RDD-id placeholders of the template, so
the message reads `The checkpoint of RDD 4 ... The checkpoint RDD is 4` and the
original RDD's id is gone, while `:199` passes `originalRDD.id` and
`newRDD.id`. The second message line still carries the diagnosis hint, so it is
not misleading, but neither variant names the checkpoint directory.
The write path goes through the new check too: `:93` and `:197` test the
same condition, so the new one throws first and the old one is only reachable
when `_num_partitions` could not be written or read back. This mismatch should
get its own error condition carrying the directory path and the two counts; the
neighboring `INVALID_CHECKPOINT_DIRECTORY` (`:85`) already carries `path`.
##########
core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala:
##########
@@ -268,6 +285,70 @@ private[spark] object ReliableCheckpointRDD extends
Logging {
}
}
+ /**
+ * Write the partition count of the checkpointed RDD to the checkpoint
directory so that
+ * a later read via [[SparkContext.checkpointFile]] can detect a truncated
directory.
+ * This is done on a best-effort basis; any exception is caught, logged and
ignored so that
+ * an inability to write the file does not prevent checkpointing. See
SPARK-58883.
+ */
+ private def writePartitionCountToCheckpointDir(
+ sc: SparkContext, partitionCount: Int, checkpointDirPath: Path): Unit = {
+ try {
+ val countFilePath = new Path(checkpointDirPath,
checkpointPartitionCountFileName())
+ val bufferSize = sc.conf.get(BUFFER_SIZE)
+ val fs = countFilePath.getFileSystem(sc.hadoopConfiguration)
+ // overwrite = false: matches _partitioner's write helper; a second
checkpoint to the
+ // same directory would fail here (caught and logged below), which is
acceptable.
+ val fileOutputStream = fs.create(countFilePath, false, bufferSize)
Review Comment:
The two comment lines above `:302` justify `overwrite = false` with "a
second checkpoint to the same directory would fail here", but that case is not
reachable. The directory is `<dir>/<UUID>/rdd-<rddId>`, the UUID changes on
every `setCheckpointDir`, rddId is unique within a SparkContext,
`writeRDDToCheckpointDirectory` has exactly one call site
(`ReliableRDDCheckpointData.scala:61`), and `RDDCheckpointData.checkpoint()`
gates on `cpState` so it runs once.
If it were reachable the conclusion would invert too: `overwrite = false`
keeps the first value, so every later read compares a stale count against a new
directory and throws a mismatch as soon as the two disagree. "Keeps this
consistent with `_partitioner`" is all the comment needs to say.
##########
core/src/test/scala/org/apache/spark/CheckpointSuite.scala:
##########
@@ -739,6 +739,92 @@ class CheckpointStorageSuite extends SparkFunSuite with
LocalSparkContext {
parameters = Map("path" -> rddPath.toString))
}
}
+
+ // SPARK-58883: truncated checkpoint directory (trailing part-* file
deleted) should be
+ // detected when reading back via SparkContext.checkpointFile.
+ test("SPARK-58883: reading a truncated checkpoint directory throws an
error") {
+ withTempDir { checkpointDir =>
+ val conf = new SparkConf().set(UI_ENABLED.key, "false")
+ sc = new SparkContext("local", "test", conf)
+ sc.setCheckpointDir(checkpointDir.toString)
+ val rdd = sc.makeRDD(1 to 20, numSlices = 4)
+ rdd.checkpoint()
+ rdd.collect()
+
+ val checkpointPath = new Path(rdd.getCheckpointFile.get)
+ val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration)
+
+ // Delete the last partition file; the remaining files are still
contiguous so the
+ // old contiguity check would pass silently and return a 3-partition RDD.
+ val lastPartFile = new Path(checkpointPath, "part-00003")
+ assert(fs.exists(lastPartFile), "expected part-00003 to exist before
deletion")
+ fs.delete(lastPartFile, false)
+
+ // Reading back should now throw because _num_partitions records the
original count.
+ // The ReliableCheckpointRDD has no separate "original" RDD on the read
path, so
+ // its own id is used for both RDD id fields.
+ val recoveredRDD = sc.checkpointFile[Int](rdd.getCheckpointFile.get)
+ checkError(
+ exception = intercept[SparkException](recoveredRDD.partitions),
+ condition = "CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH",
+ sqlState = Some("58030"),
+ parameters = Map(
+ "originalRDDId" -> recoveredRDD.id.toString,
+ "originalRDDLength" -> "4",
+ "newRDDId" -> recoveredRDD.id.toString,
+ "newRDDLength" -> "3"))
+ }
+ }
+
+ test("SPARK-58883: checkpoint directory without _num_partitions is read
without error") {
+ // Backward compatibility: a checkpoint written before SPARK-58883 has no
_num_partitions
+ // file. Removing it must not prevent the RDD from being read.
+ withTempDir { checkpointDir =>
+ val conf = new SparkConf().set(UI_ENABLED.key, "false")
+ sc = new SparkContext("local", "test", conf)
+ sc.setCheckpointDir(checkpointDir.toString)
+ val rdd = sc.makeRDD(1 to 20, numSlices = 4)
+ rdd.checkpoint()
+ rdd.collect()
+
+ val checkpointPath = new Path(rdd.getCheckpointFile.get)
+ val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration)
+
+ // Remove the metadata file to simulate a pre-SPARK-58883 checkpoint.
+ val countFile = new Path(checkpointPath, "_num_partitions")
+ fs.delete(countFile, false)
Review Comment:
Delete the write at `ReliableCheckpointRDD.scala:189` and only the first of
the three tests goes red. Test 2 deletes `_num_partitions` (`:795`) and test 3
overwrites it (`:818`), neither asserting the file exists first, and
`fs.delete` returning false is ignored while `fs.create(path, true)` on a
missing path is silent. So those two show that reading works, not the backward
compatibility and corruption tolerance their comments claim.
Test 1 does guard its precondition (`:760`), and so does an older test in
the same file (`:707`). One `assert(fs.exists(countFile))` in each of tests 2
and 3 covers it, and asserting the WARN in test 3 with `withLogAppender` is
what separates "corruption tolerated" from "the check never ran".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
