[SPARK-35784][SS] Implementation for RocksDB instance #32928

xuanyuanking · 2021-06-16T09:04:50Z

What changes were proposed in this pull request?

The implementation for the RocksDB instance, which is used in the RocksDB state store. It plays a role as a handler for the RocksDB instance and RocksDBFileManager.

Why are the changes needed?

Part of the RocksDB state store implementation.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

New UT added.

SparkQA · 2021-06-16T09:14:47Z

Test build #139867 has finished for PR 32928 at commit bbb534e.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class RocksDB(
class ByteArrayPair(var key: Array[Byte] = null, var value: Array[Byte] = null)
case class RocksDBConf(
case class AcquiredThreadInfo(

SparkQA · 2021-06-16T11:10:58Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44393/

dongjoon-hyun · 2021-06-17T23:35:40Z

sql/core/pom.xml

Shall we use the latest version?

https://mvnrepository.com/artifact/org.rocksdb/rocksdbjni/6.20.3

@dongjoon-hyun Thanks for taking care. Following the comments from Yikun in http://apache-spark-developers-list.1001551.n3.nabble.com/DISCUSS-Add-RocksDB-StateStore-tt30645.html, I planned to upgrade the rocksdb version after all the code shipped. At least >=6.4.6 for the Arm64 support.
Personally, I think it's good to keep the 6.2.2 since this is well tested in the production environment and it's safe to use it in the first stage. I believe the latest RocksDB versions are stable enough, but the state store implementation has its own design on the snapshot/checkpoint file architecture, it's good to spend more time on testing whether we can safely upgrade it to the latest RocksDB version. WDYT?

Sounds good to have well tested version first.

SparkQA · 2021-06-19T20:08:02Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44553/

SparkQA · 2021-06-19T20:41:01Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44553/

dongjoon-hyun

Thank you, @xuanyuanking . We can choose the best way for the Apache Spark community.

Just a side note: all RocksDB versions including 6.20.3 seems to be blocker for Apple Silicon Mac environment.

facebook/rocksdb#7720

In the community, there is another Spark JIRA issue considering RocksDB as a replacement for LevelDB.

https://issues.apache.org/jira/browse/SPARK-35782

cc @dbtsai , @viirya , @sunchao , @cloud-fan , @gengliangwang

dongjoon-hyun

Please fix Scala 2.13 build, @xuanyuanking .

xuanyuanking · 2021-06-21T15:56:02Z

@dongjoon-hyun Thanks for the information!

Please fix Scala 2.13 build, @xuanyuanking .

Sure

SparkQA · 2021-06-21T17:21:19Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44614/

SparkQA · 2021-06-21T17:30:17Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44614/

SparkQA · 2021-06-21T19:01:55Z

Test build #140087 has finished for PR 32928 at commit 80154b2.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class RocksDB(
class ByteArrayPair(var key: Array[Byte] = null, var value: Array[Byte] = null)
case class RocksDBConf(
case class AcquiredThreadInfo(

HeartSaVioR · 2021-06-22T07:02:54Z

I'd rather say the version bump would be a no-go till we figure out either it is backward compatible (with RocksDB 6.2.2 as it has been used for production), or quite easy way to migrate the old one to the new one. This applies to the changes on LevelDB to RocksDB as well (SPARK-35782).

Given the importance, I'd like to see the efforts in different JIRA and be orthogonal efforts with RocksDB state store provider.

viirya · 2021-06-22T06:53:55Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala

Is this basically the same as #32767, right?

If I understand correctly, commit 80154b2 should be the only change we need to review here.

Yep. We only need to review the latest commit for all the WIP pr related to the RocksDB state store.

viirya · 2021-06-22T06:57:36Z

sql/core/src/main/scala/org/apache/spark/sql/util/RocksDBLoader.scala

RocksDB -> RocksDBLoader

Done in 606a38b

viirya · 2021-06-22T06:58:43Z

sql/core/src/main/scala/org/apache/spark/sql/util/RocksDBLoader.scala

sql.util? why not in org.apache.spark.sql.execution.streaming?

It'd be nice if we can place to the package RocksDB state store provider has/will be placed, which I guess either org.apache.spark.sql.execution.streaming or org.apache.spark.sql.execution.streaming.state.

You may find the loader only related to the library loading work and not streaming only, so that's why I put it into sql.util.
Ether way is OK to me, WDYT?

RocksDB is only used for RocksDB state store provider, hence better to put it in same place. We may want to move this to other place like core if we replace LevelDB with RocksDB, but moving this doesn't break anything so no need to consider in advance.

Make sense, moved to streaming module.

viirya · 2021-06-22T07:02:45Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

Can these options be configured in the future?

The options from line 59-62 should not configurable since it's mainly for binding the sync mode of RocksDB. For the tableFormatConfig, I think they can be configured in the future if user have more performance requirements.

viirya · 2021-06-22T07:04:49Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

manager -> fileManager?

Thanks, done in f928820

viirya · 2021-06-22T07:13:53Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

If old value is not null, couldn't it be from committed db? How do we know it must be uncommitted one (and skipping increasing numUncommittedKeys)?

As the functionality of getFromBatchAndDB, it will get the key in both write batch and committed db.

This can ensure that new key appears regardless of where it comes; so looks accurate to add 1.

I see. Makes sense.

viirya · 2021-06-22T07:15:37Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

Same here, if the value could be from committed db, how can we always decrease numUncommittedKeys?

This also works as well regardless of where it comes. If it's removed from write batch, it should either from committed RocksDB (key is going to be removed) or uncommitted write batch (numUncommittedKeys should be increased in prior).

We should consider the num[Unc/C]ommittedKeys as a kind of status here, the number here means the absolute value from the beginning(version 0) to now. So that's also why there's no place we set the numUncommittedKeys to 0.

viirya · 2021-06-22T07:17:43Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

nit: don't need s"".

Thanks, done in f928820

viirya · 2021-06-22T07:19:06Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

in $checkpointDir -> "to DFS"?

Thanks, done in f928820

viirya · 2021-06-22T07:24:58Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

openDB() will close DB. This looks redundant.

We're modifying workingDir so looks safer to close DB first before modifying the dir.

Make sense, I delete the closeDB in openDB in f928820

SparkQA · 2021-06-25T12:39:24Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44854/

SparkQA · 2021-06-25T13:15:00Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44854/

SparkQA · 2021-06-25T14:09:11Z

Test build #140322 has finished for PR 32928 at commit 606a38b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR

First round.

HeartSaVioR · 2021-06-28T01:27:59Z

sql/core/src/main/scala/org/apache/spark/sql/util/RocksDBLoader.scala

It'd be nice if we can place to the package RocksDB state store provider has/will be placed, which I guess either org.apache.spark.sql.execution.streaming or org.apache.spark.sql.execution.streaming.state.

HeartSaVioR · 2021-06-28T04:26:52Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

nit: checkpointing

Thanks! Done in next commit.

HeartSaVioR · 2021-06-28T05:15:42Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

just wondering; where it is being used? this is only defined as val hence the question.

It's mainly used in the UT.

HeartSaVioR · 2021-06-28T05:35:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

We're modifying workingDir so looks safer to close DB first before modifying the dir.

HeartSaVioR · 2021-06-28T07:42:19Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+    acquireLock.notifyAll()
+  }
+
+  private def getDBProperty(property: String): Long = {


Yea, it's used in the metrics-related PR. I delete it in the next commit.

HeartSaVioR · 2021-06-28T08:03:35Z

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala

+  def withDB[T](
+      remoteDir: String,
+      version: Int = 0,
+      conf: RocksDBConf = RocksDBConf().copy(compactOnCommit = false, minVersionsToRetain = 100),


Here the conf param is always provided - is there usage on using default value?

No usage for the default value, delete the default value in the next commit.

HeartSaVioR · 2021-06-28T08:03:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+}
+
+case class AcquiredThreadInfo(
+    threadRef: WeakReference[Thread] = new WeakReference[Thread](Thread.currentThread()),


Is there any caller providing these parameters manually?

No. Make threadRef and tc as public member in the next commit.

HeartSaVioR · 2021-06-28T08:07:52Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

This can ensure that new key appears regardless of where it comes; so looks accurate to add 1.

HeartSaVioR · 2021-06-28T08:10:48Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

This also works as well regardless of where it comes. If it's removed from write batch, it should either from committed RocksDB (key is going to be removed) or uncommitted write batch (numUncommittedKeys should be increased in prior).

HeartSaVioR · 2021-06-28T08:16:58Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+  @volatile private var db: NativeRocksDB = _
+  @volatile private var loadedVersion = -1L   // -1 = nothing valid is loaded
+  @volatile private var numCommittedKeys = 0L
+  @volatile private var numUncommittedKeys = 0L


The name is actually confusing. This made me think we are tracking committed keys vs uncommitted keys (not including committed keys), but the reality is, we are tracking the number of keys in committed RocksDB, and the number of keys in (committed + uncommitted) which will be committed.

if we couldn't find a better name, we need to add code comment at least to explain this.

Agree, add more comment in the next commit.

Maybe be numKeysInCurrentVersion?

Rethink about it, I can get what it means. So it is not actually straightforward.

Yea. It's might hard to explain all the meaning in the name. So I prefer to have a long comment here 😂

xuanyuanking · 2021-06-28T12:29:37Z

All comments addressed done. Ready for another round of review.

HeartSaVioR

+1

I'll wait @viirya for a couple of days to finalize his review. I'll merge this in this Thursday if there's no further feedback.

dongjoon-hyun · 2021-06-29T02:29:06Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+    } catch {
+      case t: Throwable =>
+        loadedVersion = -1  // invalidate loaded data
+        throw t


After we re-throw this, loadedVersion is used for some purpose in the upper layer? I'm wondering if this is recoverable error and loadedVersion is a valuable information to be used later in that case. Never mind.

dongjoon-hyun · 2021-06-29T02:37:47Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+        "writeBatch" -> writeTimeMs,
+        "flush" -> flushTimeMs,
+        "compact" -> compactTimeMs,
+        "pauseBg" -> pauseTimeMs,


If you don't mind, shall we use a full name instead of Bg?

"pauseBackground" might be a little bit longer, how about change it to a simple "pause"? Actually this is only for the internal temp map key, we'll have a detailed metrics name in the later PR.

val CUSTOM_METRIC_PAUSE_TIME = StateStoreCustomTimingMetric( "rocksdbCommitPauseLatency", "RocksDB: commit - pause background time")

Changed to pause in 641a91e

dongjoon-hyun · 2021-06-29T02:47:06Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+      silentDeleteRecursively(localRootDir, "stopping")
+    } catch {
+      case e: Exception =>
+        logWarning("Error closing RocksDB", e)


If we want to do logging only at all Exception, Utils.tryLogNonFatalError might be better.

Thanks for the advice. But since NonFatal doesn't include InterruptedException which is also an Exception, maybe we'd better not change to use Utils.tryLogNonFatalError here.

dongjoon-hyun · 2021-06-29T02:56:31Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+  @volatile private var numCommittedKeys = 0L
+  // number of keys which will be committed in the next version
+  @volatile private var numUncommittedKeys = 0L
+  @volatile private var acquiredThreadInfo: AcquiredThreadInfo = _


nit. Could you add @GuardedBy("acquireLock")?

Thanks, done in the next commit.

dongjoon-hyun · 2021-06-29T02:59:13Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+    iter.seekToFirst()
+
+    // Attempt to close this iterator if there is a task failure, or a task interruption.
+    // This is a hack because it assumes that the RocksDB inside running in a task.


maybe, that the RocksDB inside running -> that the RocksDB is running?

Thanks, changed in the next commit.

viirya · 2021-06-29T03:02:03Z

Thanks @HeartSaVioR. I will take another look.

viirya · 2021-06-29T05:40:04Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+  // number of keys in all committed versions before
+  @volatile private var numCommittedKeys = 0L


why it is for "all committed versions"? In load, we just overwrite it numCommittedKeys = metadata.numKeys. So looks like it is only for the loaded version.

It's also used in the commit operation, which will be introduced in the next commit: https://github.com/apache/spark/pull/32933/files#diff-4663b8349511dc040c3d5204bb6868064002d0b45aadaf1f1795a683897a5c7dR224

Here "all" seems to make confusion. The number only represents for "one" version, hence the confusion.

Now I think it'd be nice if we can rename the field name to represent these fields are representing the number of keys per certain version. I guess below candidates could help avoiding confusion:

numKeysOnLoadedVersion - numKeysOnWritingVersion

numKeysOnCommittedVersion - numKeysOnUncommittedCommittingVersion

or better names

even without code comment.

+1 for the numKeysOnLoadedVersion - numKeysOnWritingVersion. Reviewed the new name in all the places (includes the further PRs) and it has a clear meaning. Thanks for the advice!

looks good to me.

viirya · 2021-06-29T05:49:33Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+      flushOptions.close()
+      dbOptions.close()
+      dbLogger.close()
+      silentDeleteRecursively(localRootDir, "stopping")


stopping -> closing RocksDB?

Thanks, done in the next commit.

viirya · 2021-06-29T05:53:22Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+          case InfoLogLevel.DEBUG_LEVEL => logDebug(_)
+          case _ => logTrace(_)
+        }
+        loggingFunc(s"[Native-${infoLogLevel.getValue}] $logMsg")


Native -> NativeRocksDB?

Thanks, done in the next commit.

viirya · 2021-06-29T05:53:52Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+    if (log.isDebugEnabled) dbLogLevel = InfoLogLevel.DEBUG_LEVEL
+    dbOptions.setLogger(dbLogger)
+    dbOptions.setInfoLogLevel(dbLogLevel)
+    logInfo(s"Set DB native logging level to $dbLogLevel")


We can be more specified. DB -> RocksDB

Sure, done in the next commit.

viirya · 2021-06-29T05:55:10Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala

+  // Configuration that specifies whether to compact the RocksDB data every time data is committed
+  private val COMPACT_ON_COMMIT_CONF = ConfEntry("compactOnCommit", "false")
+  private val PAUSE_BG_WORK_FOR_COMMIT_CONF = ConfEntry("pauseBackgroundWorkForCommit", "true")
+  private val BLOCK_SIZE_KB_CONF = ConfEntry("blockSizeKB", "4")
+  private val BLOCK_CACHE_SIZE_MB_CONF = ConfEntry("blockCacheSizeMB", "8")
+  private val LOCK_ACQUIRE_TIMEOUT_MS_CONF = ConfEntry("lockAcquireTimeoutMs", "60000")


Note that we should document these configs at the end.

Yea, I planned to document all the config for RocksDB in the user guide. If we have the configurable requirement in the future, people can easily be referenced.

viirya

A few minor comments.

SparkQA · 2021-06-29T22:35:13Z

Test build #140374 has finished for PR 32928 at commit 21e1728.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

viirya

lgtm. @dongjoon-hyun do you have more comments? If no, I will merge this.

dongjoon-hyun · 2021-06-30T00:36:55Z

No other comments. Thank you for asking, @viirya .

viirya · 2021-06-30T00:46:05Z

Thank you @dongjoon-hyun.

viirya · 2021-06-30T00:46:19Z

Thanks @xuanyuanking for this work! Merging to master.

xuanyuanking · 2021-06-30T02:10:15Z

Thanks all for the review! I'll rebase the other PRs soon.

github-actions bot added BUILD CORE SQL STRUCTURED STREAMING labels Jun 16, 2021

xuanyuanking mentioned this pull request Jun 16, 2021

[SPARK-35628][SS] RocksDBFileManager - load checkpoint from DFS #32767

Closed

dongjoon-hyun reviewed Jun 17, 2021

View reviewed changes

dongjoon-hyun reviewed Jun 20, 2021

View reviewed changes

xuanyuanking force-pushed the SPARK-35784 branch from bbb534e to 80154b2 Compare June 21, 2021 15:57

viirya reviewed Jun 22, 2021

View reviewed changes

xuanyuanking force-pushed the SPARK-35784 branch from 80154b2 to 606a38b Compare June 25, 2021 11:25

xuanyuanking added 3 commits June 28, 2021 15:27

Add the implementation for RocksDB instance

7903f94

address comments

263f333

address comment

f928820

xuanyuanking force-pushed the SPARK-35784 branch from 606a38b to f928820 Compare June 28, 2021 07:28

xuanyuanking changed the title ~~[WIP][SPARK-35784] Implementation for RocksDB instance~~ [SPARK-35784] Implementation for RocksDB instance Jun 28, 2021

HeartSaVioR reviewed Jun 28, 2021

View reviewed changes

address comment

72b6fe5

fix

aba142d

HeartSaVioR approved these changes Jun 29, 2021

View reviewed changes

dongjoon-hyun reviewed Jun 29, 2021

View reviewed changes

HeartSaVioR changed the title ~~[SPARK-35784] Implementation for RocksDB instance~~ [SPARK-35784][SS] Implementation for RocksDB instance Jun 29, 2021

dongjoon-hyun reviewed Jun 29, 2021

View reviewed changes

viirya reviewed Jun 29, 2021

View reviewed changes

viirya approved these changes Jun 29, 2021

View reviewed changes

xuanyuanking added 2 commits June 29, 2021 15:00

address comments

641a91e

rename

21e1728

viirya approved these changes Jun 29, 2021

View reviewed changes

viirya closed this in 3257a30 Jun 30, 2021

xuanyuanking deleted the SPARK-35784 branch June 30, 2021 02:10

		// number of keys in all committed versions before
		@volatile private var numCommittedKeys = 0L

[SPARK-35784][SS] Implementation for RocksDB instance #32928

[SPARK-35784][SS] Implementation for RocksDB instance #32928

Uh oh!

Conversation

xuanyuanking commented Jun 16, 2021

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

SparkQA commented Jun 16, 2021

Uh oh!

SparkQA commented Jun 16, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 19, 2021

Uh oh!

SparkQA commented Jun 19, 2021

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

xuanyuanking commented Jun 21, 2021

Uh oh!

SparkQA commented Jun 21, 2021

Uh oh!

SparkQA commented Jun 21, 2021

Uh oh!

SparkQA commented Jun 21, 2021

Uh oh!

HeartSaVioR commented Jun 22, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

HeartSaVioR commented Jun 22, 2021 •

edited

Loading