[GitHub] [spark] xuanyuanking commented on a change in pull request #32582: [WIP][SPARK-35436] RocksDBFileManager - save checkpoint to DFS

[GitBox <gi...@apache.org>]

xuanyuanking commented on a change in pull request #32582:
URL: https://github.com/apache/spark/pull/32582#discussion_r634466838



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
##########
@@ -0,0 +1,413 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.streaming.state
+
+import java.io.{File, FileInputStream, InputStream}
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.Files
+import java.util.UUID
+import java.util.concurrent.ConcurrentHashMap
+import java.util.zip.{ZipEntry, ZipOutputStream}
+
+import scala.collection.JavaConverters._
+import scala.collection.Seq
+
+import com.fasterxml.jackson.annotation.JsonInclude.Include
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{DefaultScalaModule, ScalaObjectMapper}
+import org.apache.commons.io.{FilenameUtils, IOUtils}
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.json4s.NoTypeHints
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.streaming.CheckpointFileManager
+import org.apache.spark.util.Utils
+
+/**
+ * Class responsible for syncing RocksDB checkpoint files from local disk to DFS.
+ * Every version checkpoint is saved in specific directory structure that allows successive
+ * versions to reuse to SST data files. This allows each commit to be incremental, only new SST
+ * files generated by RocksDB will be uploaded. The directory structures on local disk and in DFS
+ * are as follows.
+ *
+ * Local checkpoint dir structure
+ * ------------------------------
+ * RocksDB generates a bunch of files in the local checkpoint directory. The most important among
+ * them are the SST files; they are the actual log structured data files. Rest of the files contain
+ * the metadata necessary for RocksDB to read the SST files and start from the checkpoint.
+ * Note that the SST files are hard links to files in the RocksDB's working directory, and therefore
+ * successive checkpoints can share some of the SST files. So these SST files have to be copied to
+ * DFS in shared directory such that different committed versions can save them.
+ *
+ *    localCheckpointDir
+ *                  |
+ *                  +-- OPTIONS-000005
+ *                  +-- MANIFEST-000008
+ *                  +-- CURRENT
+ *                  +-- 00007.sst
+ *                  +-- 00011.sst
+ *                  +-- archive
+ *                  |     +-- 00008.log
+ *                  |     +-- 00013.log
+ *                  ...
+ *
+ *
+ * DFS directory structure after saving to DFS as version 10
+ * -----------------------------------------------------------
+ * The SST files are given unique file names and copied to the shared SST subdirectory. Every
+ * version maintains a mapping of local SST file name to the unique file name in DFS. This mapping
+ * is saved in a JSON file (named `metadata`), which is zipped along with other non-SST checkpoint
+ * files into a single file `[version].zip`.
+ *
+ *    dfsRootDir
+ *           |
+ *           +-- SSTs
+ *           |      +-- 00007-[uuid1].sst
+ *           |      +-- 00011-[uuid2].sst
+ *           +-- logs
+ *           |      +-- 00008-[uuid3].log
+ *           |      +-- 00013-[uuid4].log
+ *           +-- 10.zip
+ *           |      +-- metadata         <--- contains mapping between 00007.sst and [uuid1].sst
+ *           |      +-- OPTIONS-000005
+ *           |      +-- MANIFEST-000008
+ *           |      +-- CURRENT
+ *           |      ...
+ *           |
+ *           +-- 9.zip
+ *           +-- 8.zip
+ *           ...
+ *
+ * Note the following.
+ * - Each [version].zip is a complete description of all the data and metadata needed to recover
+ *   a RocksDB instance at the corresponding version. This is unlike the [version].delta files of
+ *   HDFSBackedStateStore where previous delta files needs to be read to be recovered.
+ * - This is safe wrt speculatively executed tasks running concurrently in different executors
+ *   as each task would upload a different copy of the generated SST files and
+ *   atomically update the [version].zip.
+ * - SST files are identified uniquely based on their file name and file size.
+ * - SST files can be reused only across adjacent checkpoints/versions.
+ * - This class is thread-safe. Specifically, it is safe to concurrently delete old files from a
+ *   different thread than the task thread saving files.
+ *
+ * @param dfsRootDir  Directory where the [version].zip files will be stored
+ * @param localTempDir Local directory for temporary work
+ * @param hadoopConf   Hadoop configuration for talking to DFS
+ * @param loggingId    Id that will be prepended in logs for isolating concurrent RocksDBs
+ */
+class RocksDBFileManager(
+  dfsRootDir: String,
+  localTempDir: File,
+  hadoopConf: Configuration,
+  loggingId: String = "")
+  extends Logging {
+
+  import RocksDBImmutableFile._
+
+  private val versionToRocksDBFiles = new ConcurrentHashMap[Long, Seq[RocksDBImmutableFile]]
+  private lazy val fm = CheckpointFileManager.create(new Path(dfsRootDir), hadoopConf)
+  private val fs = new Path(dfsRootDir).getFileSystem(hadoopConf)
+
+  /** Save all the files in given local checkpoint directory as a committed version in DFS */
+  def saveCheckpointToDfs(checkpointDir: File, version: Long, numKeys: Long): Unit = {
+    logFilesInDir(checkpointDir, s"Saving checkpoint files for version $version")
+    val allCheckpointFiles = listRocksDBFiles(checkpointDir)
+    val (localImmutableFiles, localOtherFiles) = allCheckpointFiles.partition(isImmutableFile)

Review comment:
       Per https://github.com/apache/spark/pull/32272#discussion_r632908535, we only generate the RocksDBLogFile object in the operation of saving checkpoint here.




-- 
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