You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/06/03 10:55:03 UTC
[GitHub] [spark] xuanyuanking opened a new pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
xuanyuanking opened a new pull request #32767:
URL: https://github.com/apache/spark/pull/32767
### What changes were proposed in this pull request?
The implementation for the load operation of RocksDBFileManager.
### Why are the changes needed?
Provide the functionality of loading all necessary files for specific checkpoint versions from DFS to the given local directory.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New UT added.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867750237
**[Test build #140270 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140270/testReport)** for PR 32767 at commit [`fdd0d61`](https://github.com/apache/spark/commit/fdd0d61a85c9451d3417b5a1992c22ea6c2eb13e).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656934444
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -153,6 +156,49 @@ class RocksDBFileManager(
logInfo(s"Saved checkpoint file for version $version")
}
+ /**
+ * Load all necessary files for specific checkpoint version from DFS to given local directory.
+ * If version is 0, then it will deleted all files in the directory. For other versions, it
+ * ensures that only the exact files generated during checkpointing will be present in the
+ * local directory.
+ */
+ def loadCheckpointFromDfs(version: Long, localDir: File): RocksDBCheckpointMetadata = {
+ logInfo(s"Loading checkpoint files for version $version")
+ val metadata = if (version == 0) {
+ if (localDir.exists) Utils.deleteRecursively(localDir)
+ localDir.mkdirs()
+ RocksDBCheckpointMetadata(Seq.empty, 0)
+ } else {
+ // Delete all non-immutable files in local dir, and unzip new ones from DFS commit file
+ listRocksDBFiles(localDir)._2.foreach(_.delete())
+ Utils.unzipFromFile(fs, dfsBatchZipFile(version), localDir)
+
+ // Copy the necessary immutable files
+ val metadataFile = localMetadataFile(localDir)
+ val metadata = RocksDBCheckpointMetadata.readFromFile(metadataFile)
+ logInfo(s"Read metadata for version $version:\n${metadata.prettyJson}")
+ loadImmutableFilesFromDfs(metadata.immutableFiles, localDir)
+ versionToRocksDBFiles.put(version, metadata.immutableFiles)
+ metadataFile.delete()
+ metadata
+ }
+ logFilesInDir(localDir, s"Loaded checkpoint files for version $version")
+ metadata
+ }
+
+ /** Get the latest version available in the DFS directory. If no data present, it returns 0. */
+ def getLatestVersion(): Long = {
Review comment:
Yes.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853893524
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/139284/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r658100733
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
@xuanyuanking Seems you overwrite previous change? This looks the previous version.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867747933
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44801/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868324671
**[Test build #140305 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140305/testReport)** for PR 32767 at commit [`7279d43`](https://github.com/apache/spark/commit/7279d434ddba27a93924577458f8dedf6bf340c5).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r655966166
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
Oh, I thought it is two lines.
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
yea :)
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853815518
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656946160
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -153,6 +156,49 @@ class RocksDBFileManager(
logInfo(s"Saved checkpoint file for version $version")
}
+ /**
+ * Load all necessary files for specific checkpoint version from DFS to given local directory.
+ * If version is 0, then it will deleted all files in the directory. For other versions, it
+ * ensures that only the exact files generated during checkpointing will be present in the
+ * local directory.
+ */
+ def loadCheckpointFromDfs(version: Long, localDir: File): RocksDBCheckpointMetadata = {
+ logInfo(s"Loading checkpoint files for version $version")
+ val metadata = if (version == 0) {
+ if (localDir.exists) Utils.deleteRecursively(localDir)
+ localDir.mkdirs()
+ RocksDBCheckpointMetadata(Seq.empty, 0)
+ } else {
+ // Delete all non-immutable files in local dir, and unzip new ones from DFS commit file
Review comment:
Yes. The consideration here is mainly for immutable files like sst/log files. We can avoid IO for the immutable files shared among different versions.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r658456691
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
@xuanyuanking friendly reminder
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867368334
@xuanyuanking Oh we need to fix new code conflicts as well.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866838668
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140201/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866838668
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140201/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866725363
**[Test build #140201 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140201/testReport)** for PR 32767 at commit [`7c79a29`](https://github.com/apache/spark/commit/7c79a290dccf634f059b68aaff73a46842f29021).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656918105
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
Yea, we'll have multiple lines here in the next commit :)
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r644694974
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala
##########
@@ -20,12 +20,125 @@ package org.apache.spark.sql.execution.streaming.state
import java.io._
import java.nio.charset.Charset
+import scala.language.implicitConversions
+
import org.apache.commons.io.FileUtils
+import org.apache.hadoop.conf.Configuration
import org.apache.spark._
+import org.apache.spark.sql.catalyst.util.quietly
+import org.apache.spark.sql.execution.streaming.CreateAtomicTestManager
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.util.Utils
class RocksDBSuite extends SparkFunSuite {
+ test("RocksDBFileManager: upload only new immutable files") {
+ withTempDir { dir =>
+ val dfsRootDir = dir.getAbsolutePath
+ val verificationDir = Utils.createTempDir().getAbsolutePath // local dir to load checkpoints
+ val fileManager = new RocksDBFileManager(
+ dfsRootDir, Utils.createTempDir(), new Configuration)
+ val sstDir = s"$dfsRootDir/SSTs"
+ def numRemoteSSTFiles: Int = listFiles(sstDir).length
+ val logDir = s"$dfsRootDir/logs"
+ def numRemoteLogFiles: Int = listFiles(logDir).length
+
+ // Verify behavior before any saved checkpoints
+ assert(fileManager.getLatestVersion() === 0)
+
+ // Try to load incorrect versions
+ intercept[FileNotFoundException] {
+ fileManager.loadCheckpointFromDfs(1, Utils.createTempDir())
+ }
+
+ // Save a version of checkpoint files
+ val cpFiles1 = Seq(
+ "sst-file1.sst" -> 10,
+ "sst-file2.sst" -> 20,
+ "other-file1" -> 100,
+ "other-file2" -> 200,
+ "archive/00001.log" -> 1000,
+ "archive/00002.log" -> 2000
+ )
+ saveCheckpointFiles(fileManager, cpFiles1, version = 1, numKeys = 101)
+ assert(fileManager.getLatestVersion() === 1)
+ assert(numRemoteSSTFiles == 2) // 2 sst files copied
+ assert(numRemoteLogFiles == 2) // 2 log files copied
+
+ // Load back the checkpoint files into another local dir with existing files and verify
+ generateFiles(verificationDir, Seq(
+ "sst-file1.sst" -> 11, // files with same exist but different sizes, should get overwritten
+ "other-file1" -> 101,
+ "archive/00001.log" -> 1001,
+ "random-sst-file.sst" -> 100, // unnecessary files, should get deleted
+ "random-other-file" -> 9,
+ "00005.log" -> 101,
+ "archive/00007.log" -> 101
+ ))
+ loadAndVerifyCheckpointFiles(fileManager, verificationDir, version = 1, cpFiles1, 101)
Review comment:
Per https://github.com/apache/spark/pull/32582#discussion_r642036286
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866757910
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44729/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868326321
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140305/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r655927681
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -153,6 +156,49 @@ class RocksDBFileManager(
logInfo(s"Saved checkpoint file for version $version")
}
+ /**
+ * Load all necessary files for specific checkpoint version from DFS to given local directory.
+ * If version is 0, then it will deleted all files in the directory. For other versions, it
+ * ensures that only the exact files generated during checkpointing will be present in the
+ * local directory.
+ */
+ def loadCheckpointFromDfs(version: Long, localDir: File): RocksDBCheckpointMetadata = {
+ logInfo(s"Loading checkpoint files for version $version")
+ val metadata = if (version == 0) {
+ if (localDir.exists) Utils.deleteRecursively(localDir)
+ localDir.mkdirs()
+ RocksDBCheckpointMetadata(Seq.empty, 0)
+ } else {
+ // Delete all non-immutable files in local dir, and unzip new ones from DFS commit file
+ listRocksDBFiles(localDir)._2.foreach(_.delete())
+ Utils.unzipFromFile(fs, dfsBatchZipFile(version), localDir)
+
+ // Copy the necessary immutable files
+ val metadataFile = localMetadataFile(localDir)
+ val metadata = RocksDBCheckpointMetadata.readFromFile(metadataFile)
+ logInfo(s"Read metadata for version $version:\n${metadata.prettyJson}")
+ loadImmutableFilesFromDfs(metadata.immutableFiles, localDir)
+ versionToRocksDBFiles.put(version, metadata.immutableFiles)
+ metadataFile.delete()
+ metadata
+ }
+ logFilesInDir(localDir, s"Loaded checkpoint files for version $version")
+ metadata
+ }
+
+ /** Get the latest version available in the DFS directory. If no data present, it returns 0. */
+ def getLatestVersion(): Long = {
Review comment:
Is this only used by tests?
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r655924518
##########
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")
+ val expectedSize = file.sizeBytes
+ if (localFileSize != expectedSize) {
+ throw new IllegalStateException(
+ s"Copied $dfsFile to $localFile," +
+ s" expected $expectedSize bytes, found $localFileSize bytes ")
+ }
+ } else {
+ filesReused += 1
+ }
+ }
+ logInfo(s"Copied $filesCopied files ($bytesCopied bytes) from DFS to local.")
Review comment:
Add `filesReused` into this log message?
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
+ def unzipFromFile(fs: FileSystem, dfsZipFile: Path, localDir: File): Seq[File] = {
+ val files = new mutable.ArrayBuffer[File]()
+ val in = new ZipInputStream(fs.open(dfsZipFile))
+ var out: OutputStream = null
+ try {
+ var entry = in.getNextEntry()
+ while (entry != null) {
+ if (!entry.isDirectory) {
Review comment:
We don't process directory. Could you also mention it in the method doc?
##########
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")
+ }
Review comment:
Or just a safer guard?
##########
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")
+ }
Review comment:
When it is possible to have existing file in local dir which has same file name but not the same file in DFS?
##########
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:
We can do this logInfo after the file size check.
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
+ def unzipFromFile(fs: FileSystem, dfsZipFile: Path, localDir: File): Seq[File] = {
+ val files = new mutable.ArrayBuffer[File]()
+ val in = new ZipInputStream(fs.open(dfsZipFile))
+ var out: OutputStream = null
+ try {
+ var entry = in.getNextEntry()
+ while (entry != null) {
+ if (!entry.isDirectory) {
+ val fileName = localDir.toPath.resolve(entry.getName).getFileName.toString
+ val outFile = new File(localDir, fileName)
+ files += outFile
+ out = new FileOutputStream(outFile)
+ IOUtils.copy(in, out)
+ out.close()
+ in.closeEntry()
+ }
+ entry = in.getNextEntry()
+ }
+ in.close() // so that any error in closing does not get ignored
+ logInfo(s"Unzipped from $dfsZipFile\n\t${files.mkString("\n\t")}")
+ } finally {
+ // Close everything no matter what happened
+ IOUtils.closeQuietly(in)
+ IOUtils.closeQuietly(out)
+ }
Review comment:
Hmm, are we sure we don't need to process any error during unzipping?
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
Looks like this is not the Java doc style we follow.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -153,6 +156,49 @@ class RocksDBFileManager(
logInfo(s"Saved checkpoint file for version $version")
}
+ /**
+ * Load all necessary files for specific checkpoint version from DFS to given local directory.
+ * If version is 0, then it will deleted all files in the directory. For other versions, it
+ * ensures that only the exact files generated during checkpointing will be present in the
+ * local directory.
+ */
+ def loadCheckpointFromDfs(version: Long, localDir: File): RocksDBCheckpointMetadata = {
+ logInfo(s"Loading checkpoint files for version $version")
+ val metadata = if (version == 0) {
+ if (localDir.exists) Utils.deleteRecursively(localDir)
+ localDir.mkdirs()
+ RocksDBCheckpointMetadata(Seq.empty, 0)
+ } else {
+ // Delete all non-immutable files in local dir, and unzip new ones from DFS commit file
+ listRocksDBFiles(localDir)._2.foreach(_.delete())
+ Utils.unzipFromFile(fs, dfsBatchZipFile(version), localDir)
+
+ // Copy the necessary immutable files
+ val metadataFile = localMetadataFile(localDir)
+ val metadata = RocksDBCheckpointMetadata.readFromFile(metadataFile)
+ logInfo(s"Read metadata for version $version:\n${metadata.prettyJson}")
+ loadImmutableFilesFromDfs(metadata.immutableFiles, localDir)
+ versionToRocksDBFiles.put(version, metadata.immutableFiles)
+ metadataFile.delete()
+ metadata
+ }
+ logFilesInDir(localDir, s"Loaded checkpoint files for version $version")
+ metadata
+ }
+
+ /** Get the latest version available in the DFS directory. If no data present, it returns 0. */
+ def getLatestVersion(): Long = {
Review comment:
Is this only used by tests?
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866725363
**[Test build #140201 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140201/testReport)** for PR 32767 at commit [`7c79a29`](https://github.com/apache/spark/commit/7c79a290dccf634f059b68aaff73a46842f29021).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866818226
**[Test build #140201 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140201/testReport)** for PR 32767 at commit [`7c79a29`](https://github.com/apache/spark/commit/7c79a290dccf634f059b68aaff73a46842f29021).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867742012
**[Test build #140270 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140270/testReport)** for PR 32767 at commit [`fdd0d61`](https://github.com/apache/spark/commit/fdd0d61a85c9451d3417b5a1992c22ea6c2eb13e).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r658484168
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
……Sorry I missed this... Updating
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857562766
**[Test build #139555 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139555/testReport)** for PR 32767 at commit [`10d11b3`](https://github.com/apache/spark/commit/10d11b3036282f4a56e5f7c90e0cdf1d8a84de63).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868221000
**[Test build #140305 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140305/testReport)** for PR 32767 at commit [`7279d43`](https://github.com/apache/spark/commit/7279d434ddba27a93924577458f8dedf6bf340c5).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868218107
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44831/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r658040266
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala
##########
@@ -53,10 +62,24 @@ class RocksDBSuite extends SparkFunSuite {
"archive/00002.log" -> 2000
)
saveCheckpointFiles(fileManager, cpFiles1, version = 1, numKeys = 101)
+ assert(fileManager.getLatestVersion() === 1)
assert(numRemoteSSTFiles == 2) // 2 sst files copied
assert(numRemoteLogFiles == 2) // 2 log files copied
- // Save SAME version again with different checkpoint files and verify
+ // Load back the checkpoint files into another local dir with existing files and verify
+ generateFiles(verificationDir, Seq(
+ "sst-file1.sst" -> 11, // files with same exist but different sizes, should get overwritten
Review comment:
Ah thanks! Fix done.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857613834
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44081/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853815518
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868262279
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867750388
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140270/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868261445
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44836/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656920324
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
+ def unzipFromFile(fs: FileSystem, dfsZipFile: Path, localDir: File): Seq[File] = {
+ val files = new mutable.ArrayBuffer[File]()
+ val in = new ZipInputStream(fs.open(dfsZipFile))
+ var out: OutputStream = null
+ try {
+ var entry = in.getNextEntry()
+ while (entry != null) {
+ if (!entry.isDirectory) {
Review comment:
Make sense, method name changed and comment added.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656929673
##########
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")
+ }
Review comment:
A safer guard for checking both file names and file size.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868373147
Thanks @xuanyuanking for the contribution! I merged into master.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656934779
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -153,6 +156,49 @@ class RocksDBFileManager(
logInfo(s"Saved checkpoint file for version $version")
}
+ /**
+ * Load all necessary files for specific checkpoint version from DFS to given local directory.
+ * If version is 0, then it will deleted all files in the directory. For other versions, it
Review comment:
Thanks, done in the next commit
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857577820
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/139555/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r655951825
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -153,6 +156,49 @@ class RocksDBFileManager(
logInfo(s"Saved checkpoint file for version $version")
}
+ /**
+ * Load all necessary files for specific checkpoint version from DFS to given local directory.
+ * If version is 0, then it will deleted all files in the directory. For other versions, it
+ * ensures that only the exact files generated during checkpointing will be present in the
+ * local directory.
+ */
+ def loadCheckpointFromDfs(version: Long, localDir: File): RocksDBCheckpointMetadata = {
+ logInfo(s"Loading checkpoint files for version $version")
+ val metadata = if (version == 0) {
+ if (localDir.exists) Utils.deleteRecursively(localDir)
+ localDir.mkdirs()
+ RocksDBCheckpointMetadata(Seq.empty, 0)
+ } else {
+ // Delete all non-immutable files in local dir, and unzip new ones from DFS commit file
Review comment:
What if we just remove the all files in localDir? Just would like to know the reason we don't clear the directory but just remove the specific files. Would we need to leverage some remaining files?
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
We have been using one-liner java doc; e.g. https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala#L50
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
+ def unzipFromFile(fs: FileSystem, dfsZipFile: Path, localDir: File): Seq[File] = {
+ val files = new mutable.ArrayBuffer[File]()
+ val in = new ZipInputStream(fs.open(dfsZipFile))
+ var out: OutputStream = null
+ try {
+ var entry = in.getNextEntry()
+ while (entry != null) {
+ if (!entry.isDirectory) {
Review comment:
It'd be ideal if we make it clear in method name, like `unzipFilesFromFile`. (Ideally I'd like to see this also extracts the directory, but let's postpone it till necessary.)
In general we expect unzipping will extract the directories as well. That said, we need to make the behavior very clear to the caller side. I agree this should be mentioned to the java doc, but method name should be also intuitive to expect the actual behavior.
##########
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:
More specifically, we can do the file size check just after copyToLocalFile, and accumulations can be placed later.
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
+ def unzipFromFile(fs: FileSystem, dfsZipFile: Path, localDir: File): Seq[File] = {
+ val files = new mutable.ArrayBuffer[File]()
+ val in = new ZipInputStream(fs.open(dfsZipFile))
+ var out: OutputStream = null
+ try {
+ var entry = in.getNextEntry()
+ while (entry != null) {
+ if (!entry.isDirectory) {
+ val fileName = localDir.toPath.resolve(entry.getName).getFileName.toString
+ val outFile = new File(localDir, fileName)
+ files += outFile
+ out = new FileOutputStream(outFile)
+ IOUtils.copy(in, out)
+ out.close()
+ in.closeEntry()
+ }
+ entry = in.getNextEntry()
+ }
+ in.close() // so that any error in closing does not get ignored
+ logInfo(s"Unzipped from $dfsZipFile\n\t${files.mkString("\n\t")}")
+ } finally {
+ // Close everything no matter what happened
+ IOUtils.closeQuietly(in)
+ IOUtils.closeQuietly(out)
+ }
Review comment:
This looks safe; if there's an exception we may see some files being extracted and the one of output files may be broken, but callers will catch an exception and indicate the output directory is not healthy. If necessary let's document this in javadoc as well.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -153,6 +156,49 @@ class RocksDBFileManager(
logInfo(s"Saved checkpoint file for version $version")
}
+ /**
+ * Load all necessary files for specific checkpoint version from DFS to given local directory.
+ * If version is 0, then it will deleted all files in the directory. For other versions, it
Review comment:
nit: deleted -> delete
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
But we'd like to see this multiple lines as below review comments :)
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868435398
Great thanks @HeartSaVioR and @Ngone51. Submitted https://github.com/apache/spark/pull/33084.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868374852
Please also rebase the next PRs into master branch; we can continue reviewing next PR.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-858191734
@viirya Great thanks for your help!
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857619130
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44081/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853783621
**[Test build #139284 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139284/testReport)** for PR 32767 at commit [`ac47618`](https://github.com/apache/spark/commit/ac476189649978280487c6e32a283878af2f774f).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868260410
**[Test build #140300 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140300/testReport)** for PR 32767 at commit [`be15054`](https://github.com/apache/spark/commit/be15054b732f9b844e0841f3eee793f229c35e71).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853815518
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43808/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853783621
**[Test build #139284 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139284/testReport)** for PR 32767 at commit [`ac47618`](https://github.com/apache/spark/commit/ac476189649978280487c6e32a283878af2f774f).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r655924518
##########
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")
+ val expectedSize = file.sizeBytes
+ if (localFileSize != expectedSize) {
+ throw new IllegalStateException(
+ s"Copied $dfsFile to $localFile," +
+ s" expected $expectedSize bytes, found $localFileSize bytes ")
+ }
+ } else {
+ filesReused += 1
+ }
+ }
+ logInfo(s"Copied $filesCopied files ($bytesCopied bytes) from DFS to local.")
Review comment:
Add `filesReused` into this log message?
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
+ def unzipFromFile(fs: FileSystem, dfsZipFile: Path, localDir: File): Seq[File] = {
+ val files = new mutable.ArrayBuffer[File]()
+ val in = new ZipInputStream(fs.open(dfsZipFile))
+ var out: OutputStream = null
+ try {
+ var entry = in.getNextEntry()
+ while (entry != null) {
+ if (!entry.isDirectory) {
Review comment:
We don't process directory. Could you also mention it in the method doc?
##########
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")
+ }
Review comment:
Or just a safer guard?
##########
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")
+ }
Review comment:
When it is possible to have existing file in local dir which has same file name but not the same file in DFS?
##########
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:
We can do this logInfo after the file size check.
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
+ def unzipFromFile(fs: FileSystem, dfsZipFile: Path, localDir: File): Seq[File] = {
+ val files = new mutable.ArrayBuffer[File]()
+ val in = new ZipInputStream(fs.open(dfsZipFile))
+ var out: OutputStream = null
+ try {
+ var entry = in.getNextEntry()
+ while (entry != null) {
+ if (!entry.isDirectory) {
+ val fileName = localDir.toPath.resolve(entry.getName).getFileName.toString
+ val outFile = new File(localDir, fileName)
+ files += outFile
+ out = new FileOutputStream(outFile)
+ IOUtils.copy(in, out)
+ out.close()
+ in.closeEntry()
+ }
+ entry = in.getNextEntry()
+ }
+ in.close() // so that any error in closing does not get ignored
+ logInfo(s"Unzipped from $dfsZipFile\n\t${files.mkString("\n\t")}")
+ } finally {
+ // Close everything no matter what happened
+ IOUtils.closeQuietly(in)
+ IOUtils.closeQuietly(out)
+ }
Review comment:
Hmm, are we sure we don't need to process any error during unzipping?
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
Review comment:
Looks like this is not the Java doc style we follow.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867750388
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140270/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857619130
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44081/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866758535
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44729/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853815518
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43808/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868262279
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857619105
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44081/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853783621
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857577820
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/139555/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857454764
Rebased this PR based on #32582. It's ready for review now. cc @viirya and @HeartSaVioR
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656921453
##########
File path: core/src/main/scala/org/apache/spark/util/Utils.scala
##########
@@ -3084,6 +3085,35 @@ private[spark] object Utils extends Logging {
conf.get(config.STORAGE_BLOCKMANAGER_HEARTBEAT_TIMEOUT)
.getOrElse(Utils.timeStringAsMs(s"${conf.get(Network.NETWORK_TIMEOUT)}s"))
}
+
+ /** Decompress a zip file into a local dir. File names are read from the zip file. */
+ def unzipFromFile(fs: FileSystem, dfsZipFile: Path, localDir: File): Seq[File] = {
+ val files = new mutable.ArrayBuffer[File]()
+ val in = new ZipInputStream(fs.open(dfsZipFile))
+ var out: OutputStream = null
+ try {
+ var entry = in.getNextEntry()
+ while (entry != null) {
+ if (!entry.isDirectory) {
+ val fileName = localDir.toPath.resolve(entry.getName).getFileName.toString
+ val outFile = new File(localDir, fileName)
+ files += outFile
+ out = new FileOutputStream(outFile)
+ IOUtils.copy(in, out)
+ out.close()
+ in.closeEntry()
+ }
+ entry = in.getNextEntry()
+ }
+ in.close() // so that any error in closing does not get ignored
+ logInfo(s"Unzipped from $dfsZipFile\n\t${files.mkString("\n\t")}")
+ } finally {
+ // Close everything no matter what happened
+ IOUtils.closeQuietly(in)
+ IOUtils.closeQuietly(out)
+ }
Review comment:
Yes, we rely on the caller side to address any exceptions. Javadoc added as well.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868179234
**[Test build #140300 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140300/testReport)** for PR 32767 at commit [`be15054`](https://github.com/apache/spark/commit/be15054b732f9b844e0841f3eee793f229c35e71).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868206855
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44831/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857456948
**[Test build #139555 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139555/testReport)** for PR 32767 at commit [`10d11b3`](https://github.com/apache/spark/commit/10d11b3036282f4a56e5f7c90e0cdf1d8a84de63).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r657659814
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala
##########
@@ -53,10 +62,24 @@ class RocksDBSuite extends SparkFunSuite {
"archive/00002.log" -> 2000
)
saveCheckpointFiles(fileManager, cpFiles1, version = 1, numKeys = 101)
+ assert(fileManager.getLatestVersion() === 1)
assert(numRemoteSSTFiles == 2) // 2 sst files copied
assert(numRemoteLogFiles == 2) // 2 log files copied
- // Save SAME version again with different checkpoint files and verify
+ // Load back the checkpoint files into another local dir with existing files and verify
+ generateFiles(verificationDir, Seq(
+ "sst-file1.sst" -> 11, // files with same exist but different sizes, should get overwritten
Review comment:
files with same? with same filename?
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867747955
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44801/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866758535
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44729/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867742012
**[Test build #140270 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140270/testReport)** for PR 32767 at commit [`fdd0d61`](https://github.com/apache/spark/commit/fdd0d61a85c9451d3417b5a1992c22ea6c2eb13e).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r656932801
##########
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")
+ val expectedSize = file.sizeBytes
+ if (localFileSize != expectedSize) {
+ throw new IllegalStateException(
+ s"Copied $dfsFile to $localFile," +
+ s" expected $expectedSize bytes, found $localFileSize bytes ")
+ }
+ } else {
+ filesReused += 1
+ }
+ }
+ logInfo(s"Copied $filesCopied files ($bytesCopied bytes) from DFS to local.")
Review comment:
Thanks! Done in the next commit.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32767:
URL: https://github.com/apache/spark/pull/32767#discussion_r644694974
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala
##########
@@ -20,12 +20,125 @@ package org.apache.spark.sql.execution.streaming.state
import java.io._
import java.nio.charset.Charset
+import scala.language.implicitConversions
+
import org.apache.commons.io.FileUtils
+import org.apache.hadoop.conf.Configuration
import org.apache.spark._
+import org.apache.spark.sql.catalyst.util.quietly
+import org.apache.spark.sql.execution.streaming.CreateAtomicTestManager
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.util.Utils
class RocksDBSuite extends SparkFunSuite {
+ test("RocksDBFileManager: upload only new immutable files") {
+ withTempDir { dir =>
+ val dfsRootDir = dir.getAbsolutePath
+ val verificationDir = Utils.createTempDir().getAbsolutePath // local dir to load checkpoints
+ val fileManager = new RocksDBFileManager(
+ dfsRootDir, Utils.createTempDir(), new Configuration)
+ val sstDir = s"$dfsRootDir/SSTs"
+ def numRemoteSSTFiles: Int = listFiles(sstDir).length
+ val logDir = s"$dfsRootDir/logs"
+ def numRemoteLogFiles: Int = listFiles(logDir).length
+
+ // Verify behavior before any saved checkpoints
+ assert(fileManager.getLatestVersion() === 0)
+
+ // Try to load incorrect versions
+ intercept[FileNotFoundException] {
+ fileManager.loadCheckpointFromDfs(1, Utils.createTempDir())
+ }
+
+ // Save a version of checkpoint files
+ val cpFiles1 = Seq(
+ "sst-file1.sst" -> 10,
+ "sst-file2.sst" -> 20,
+ "other-file1" -> 100,
+ "other-file2" -> 200,
+ "archive/00001.log" -> 1000,
+ "archive/00002.log" -> 2000
+ )
+ saveCheckpointFiles(fileManager, cpFiles1, version = 1, numKeys = 101)
+ assert(fileManager.getLatestVersion() === 1)
+ assert(numRemoteSSTFiles == 2) // 2 sst files copied
+ assert(numRemoteLogFiles == 2) // 2 log files copied
+
+ // Load back the checkpoint files into another local dir with existing files and verify
+ generateFiles(verificationDir, Seq(
+ "sst-file1.sst" -> 11, // files with same exist but different sizes, should get overwritten
+ "other-file1" -> 101,
+ "archive/00001.log" -> 1001,
+ "random-sst-file.sst" -> 100, // unnecessary files, should get deleted
+ "random-other-file" -> 9,
+ "00005.log" -> 101,
+ "archive/00007.log" -> 101
+ ))
+ loadAndVerifyCheckpointFiles(fileManager, verificationDir, version = 1, cpFiles1, 101)
Review comment:
Per https://github.com/apache/spark/pull/32582#discussion_r642036286
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867720151
Thanks for the review!
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853893524
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/139284/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR closed pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
HeartSaVioR closed pull request #32767:
URL: https://github.com/apache/spark/pull/32767
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857456948
**[Test build #139555 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139555/testReport)** for PR 32767 at commit [`10d11b3`](https://github.com/apache/spark/commit/10d11b3036282f4a56e5f7c90e0cdf1d8a84de63).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866751029
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44729/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868196317
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44831/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853783621
**[Test build #139284 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139284/testReport)** for PR 32767 at commit [`ac47618`](https://github.com/apache/spark/commit/ac476189649978280487c6e32a283878af2f774f).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867747955
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44801/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868221000
**[Test build #140305 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140305/testReport)** for PR 32767 at commit [`7279d43`](https://github.com/apache/spark/commit/7279d434ddba27a93924577458f8dedf6bf340c5).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868371947
Jenkins passed. Thanks! Merging to master.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868326321
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140305/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868431278
UPDATE: we found a consistent break on Scala 2.13 build caused by this. @xuanyuanking is working on the fix so please allow us some time to fix it as follow-up PR instead of reverting this.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868377947
Copy that. I'm rebasing now. Thanks for the help!
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853805363
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43808/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-857463805
Thank you @xuanyuanking. I'll find some time to review this.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32767: [WIP][SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-853869221
**[Test build #139284 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139284/testReport)** for PR 32767 at commit [`ac47618`](https://github.com/apache/spark/commit/ac476189649978280487c6e32a283878af2f774f).
* This patch passes all tests.
* This patch **does not merge cleanly**.
* This patch adds no public classes.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868218107
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44831/
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-868179234
**[Test build #140300 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140300/testReport)** for PR 32767 at commit [`be15054`](https://github.com/apache/spark/commit/be15054b732f9b844e0841f3eee793f229c35e71).
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-866703114
Great thanks for your detailed review, @viirya @HeartSaVioR. All comments addressed.
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #32767: [SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #32767:
URL: https://github.com/apache/spark/pull/32767#issuecomment-867374452
@xuanyuanking Could you resolve the conflict and the minor comment? Then we can move this forward. Thanks!
--
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org