xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656930705
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -200,6 +246,55 @@ class RocksDBFileManager(
immutableFiles
}
+ /**
+ * Copy files from DFS directory to a local directory. It will figure out
which
+ * existing files are needed, and accordingly, unnecessary SST files are
deleted while
+ * necessary and non-existing files are copied from DFS.
+ */
+ private def loadImmutableFilesFromDfs(
+ immutableFiles: Seq[RocksDBImmutableFile], localDir: File): Unit = {
+ val requiredFileNameToFileDetails = immutableFiles.map(f =>
f.localFileName -> f).toMap
+ // Delete unnecessary local immutable files
+ listRocksDBFiles(localDir)._1
+ .foreach { existingFile =>
+ val isSameFile =
+
requiredFileNameToFileDetails.get(existingFile.getName).exists(_.isSameFile(existingFile))
+ if (!isSameFile) {
+ existingFile.delete()
+ logInfo(s"Deleted local file $existingFile")
+ }
+ }
+
+ var filesCopied = 0L
+ var bytesCopied = 0L
+ var filesReused = 0L
+ immutableFiles.foreach { file =>
+ val localFileName = file.localFileName
+ val localFile = localFilePath(localDir, localFileName)
+ if (!localFile.exists) {
+ val dfsFile = dfsFilePath(file.dfsFileName)
+ // Note: The implementation of copyToLocalFile() closes the output
stream when there is
+ // any exception while copying. So this may generate partial files on
DFS. But that is
+ // okay because until the main [version].zip file is written, those
partial files are
+ // not going to be used at all. Eventually these files should get
cleared.
+ fs.copyToLocalFile(dfsFile, new Path(localFile.getAbsoluteFile.toURI))
+ val localFileSize = localFile.length()
+ filesCopied += 1
+ bytesCopied += localFileSize
+ logInfo(s"Copied $dfsFile to $localFile - $localFileSize bytes")
Review comment:
Thanks, do the size check right after copyToLocalFile and place the
logInfo in the end.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
