anshulbaliga7 commented on code in PR #58167:
URL: https://github.com/apache/spark/pull/58167#discussion_r3862071091
##########
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:
Non-atomic write. Rewrote to `DataOutputStream.writeInt` behind a 1-byte
format version, written to a temp path and renamed into place. A rename failure
is also logged (with cause) rather than silently dropped, so every path that
disables the check now leaves a trace.
--
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]
