[SPARK-35988][SS] The implementation for RocksDBStateStoreProvider by xuanyuanking · Pull Request #33187 · apache/spark

xuanyuanking · 2021-07-02T09:48:34Z

What changes were proposed in this pull request?

Add the implementation for the RocksDBStateStoreProvider. It's the subclass of StateStoreProvider that leverages all the functionalities implemented in the RocksDB instance.

Why are the changes needed?

The interface for the end-user to use the RocksDB state store.

Does this PR introduce any user-facing change?

Yes. New RocksDBStateStore can be used in their applications.

How was this patch tested?

New UT added.

SparkQA · 2021-07-02T09:59:32Z

Test build #140577 has finished for PR 33187 at commit af24bd9.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class RocksDBStateStore(val lastVersion: Long) extends StateStore
class StateEncoder

SparkQA · 2021-07-02T10:04:58Z

Kubernetes integration test unable to build dist.

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

SparkQA · 2021-07-02T13:49:44Z

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

SparkQA · 2021-07-02T14:24:56Z

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

SparkQA · 2021-07-02T16:32:27Z

Test build #140587 has finished for PR 33187 at commit f59a9e9.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR

Looks OK, but I'd like to make sure @viirya is also OK with the change, as I'm familiar with this change and might be biased.

HeartSaVioR · 2021-07-05T02:16:59Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

This can be referenced by version so no need to be val.

Thanks, done in 9ddbf58

HeartSaVioR · 2021-07-05T02:45:57Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

I thought the characters (=,()) being used in storeIdStr are odd, but this works in MacOS and I think it would also work in Linux file systems, so OK.

Yes, the dir contains =() works in the Linux file system.

HeartSaVioR · 2021-07-05T02:52:01Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

This simply works as there's only one version (version 0) - we'll need to check the version when we add a new state encoding version.

Yea, agree. For the new encoding version, we should have branches here for different versions.

HeartSaVioR · 2021-07-05T02:53:49Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

We can deduplicate here; I can deal with it in prefix scan for RocksDB state store as I'll bring broader change for state encoding.

HeartSaVioR · 2021-07-05T02:56:31Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

nit: remove one of two empty lines

Thanks, done in 9ddbf58

HeartSaVioR · 2021-07-06T02:29:35Z

NOTE: This might have some post-review comments from #32934 .

HeartSaVioR · 2021-07-06T05:14:22Z

Finally we only have this one for RocksDB state store provider! Please rebase this so that we can continue.

xuanyuanking · 2021-07-06T06:17:08Z

Yes, finally!
Addressed the comments and rebased the code.

SparkQA · 2021-07-06T07:42:39Z

Kubernetes integration test unable to build dist.

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

SparkQA · 2021-07-06T09:52:28Z

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

SparkQA · 2021-07-06T09:57:08Z

Test build #140685 has finished for PR 33187 at commit 9ddbf58.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class RocksDBStateStore(lastVersion: Long) extends StateStore

SparkQA · 2021-07-06T10:24:18Z

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

SparkQA · 2021-07-06T10:38:12Z

Test build #140697 has finished for PR 33187 at commit a886567.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-07-06T14:44:21Z

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

SparkQA · 2021-07-06T15:18:58Z

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

SparkQA · 2021-07-06T18:19:24Z

Test build #140707 has finished for PR 33187 at commit e63a380.

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

HeartSaVioR

The code change looks good to me. Great!

One thing we haven't been addressed yet is documentation. End users have no idea how to use this provider. Probably we should introduce RocksDB state store provider in SS guide doc.

It's OK if you prefer to file a "blocker" JIRA issue to address documentation separately. I can approve this one then.

HeartSaVioR · 2021-07-08T04:12:23Z

...e/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreSuite.scala

+    assert(store.hasCommitted)
+    val storeMetrics = store.metrics
+    assert(storeMetrics.numKeys === 1)
+    assert(getCustomMetric(storeMetrics, CUSTOM_METRIC_FILES_COPIED) == 1L)


nit: I guess we can loose the condition via > 0L. "The number of files is 1" doesn't seem to something we need to check.

Agree, will update in the next commit.

HeartSaVioR · 2021-07-08T04:22:34Z

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

    assert(getSizeOfStateForCurrentVersion(store.metrics) > noDataMemoryUsed)
  }

+  test("maintenance") {


Just a head-up: "maintenance" and "snapshotting" are moved from StateStoreSuiteBase to StateStoreSuite as these tests cannot be applied to RocksDB state store provider.

Yes, since these tests have the assumption that the provider have a baseDir

xuanyuanking · 2021-07-08T06:21:19Z

One thing we haven't been addressed yet is documentation. End users have no idea how to use this provider. Probably we should introduce RocksDB state store provider in SS guide doc.

It's OK if you prefer to file a "blocker" JIRA issue to address documentation separately. I can approve this one then.

Sure, create the blocker issue SPARK-36041 for tracking the documentation.

HeartSaVioR

+1

I'll merge once the test passes.

My apologize on not waiting for reviewing on others, we are far behind on progress (compared to the planned date on RC) and this PR is somehow a blocker for session window stuff. The code has been running on production for years so it won't make some problems, but we are open to deal with post-reviews during QA phase.

SparkQA · 2021-07-08T07:03:49Z

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

SparkQA · 2021-07-08T07:36:54Z

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

viirya · 2021-07-08T07:38:55Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

+
+    override def id: StateStoreId = RocksDBStateStoreProvider.this.stateStoreId
+
+    override def version: Long = lastVersion


I recommend to be consistent with HDFSBackedStateStore where it uses version and newVersion.

Actually for RocksDB, we keep the name newVersion. The major difference here is we have a loadedVersion concept, so that's why for the provider side, we have a latestVersion.

viirya · 2021-07-08T07:41:35Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

+      verify(state == UPDATING, "Cannot put after already committed or aborted")
+      verify(key != null, "Key cannot be null")
+      require(value != null, "Cannot put a null value")
+      logDebug(s"Storing $key => $value")


Not sure if this log debug is necessary. Especially key/value are unsafe rows.

Make sense, let me delete this debug log.

viirya · 2021-07-08T07:53:01Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

+
+  /**
+   * Encodes/decodes UnsafeRows to versioned byte arrays.
+   * It uses the first byte of the generated byte array to store the version the describes how the


I guess this is a typo. "the" -> "that".

Ah yea, thanks!

viirya · 2021-07-08T07:59:19Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

+          Platform.BYTE_ARRAY_OFFSET + STATE_ENCODING_NUM_VERSION_BYTES,
+          keyBytes.length - STATE_ENCODING_NUM_VERSION_BYTES)
+        keyRow
+      } else null


nit: I rarely see style like this in Spark.

Make sense, done in the next commit.

viirya · 2021-07-08T08:01:17Z

...rc/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala

+  val CUSTOM_METRIC_CHECKPOINT_TIME = StateStoreCustomTimingMetric(
+    "rocksdbCommitCheckpointLatency", "RocksDB: commit - checkpoint time")
+  val CUSTOM_METRIC_FILESYNC_TIME = StateStoreCustomTimingMetric(
+    "rocksdbCommitPauseBgTime", "RocksDB: commit - file sync time")


rocksdbCommitPauseBgTime? Or FileSyncTime?

Yea, changed to rocksdbFileSyncTime

viirya · 2021-07-08T08:03:35Z

...e/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreSuite.scala

+      val rocksDBConfInTask: RocksDBConf = testRDD.mapPartitionsWithStateStore[RocksDBConf](
+        spark.sqlContext, testStateInfo, testSchema, testSchema, None) {
+          (store: StateStore, _: Iterator[String]) =>
+            // Use reflection to get RockDB instance


nit: RockDB -> RocksDB

Thanks, done in the next commit.

viirya

Looks good. Some minor comments. Feel free to address them in follow-up if you prefer to get this in first.

HeartSaVioR · 2021-07-08T08:12:40Z

I'm OK to wait for addressing these comments - I guess we can still merge this in this week, then it's OK for me.

I'm already working on rebasing #33038 on top of this PR. Looks like I need to bring another bunch of lines in #33038 to address RocksDB state store.

SparkQA · 2021-07-08T10:47:46Z

Kubernetes integration test unable to build dist.

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

SparkQA · 2021-07-08T11:25:05Z

Test build #140779 has finished for PR 33187 at commit 1ac5835.

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

HeartSaVioR

+1

Thanks! Merging to master/3.2!

### What changes were proposed in this pull request? Add the implementation for the RocksDBStateStoreProvider. It's the subclass of StateStoreProvider that leverages all the functionalities implemented in the RocksDB instance. ### Why are the changes needed? The interface for the end-user to use the RocksDB state store. ### Does this PR introduce _any_ user-facing change? Yes. New RocksDBStateStore can be used in their applications. ### How was this patch tested? New UT added. Closes #33187 from xuanyuanking/SPARK-35988. Authored-by: Yuanjian Li <yuanjian.li@databricks.com> Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com> (cherry picked from commit 0621e78) Signed-off-by: Jungtaek Lim <kabhwan.opensource@gmail.com>

HeartSaVioR · 2021-07-08T12:04:14Z

Thanks @xuanyuanking for contribution! Merged to master/3.2 branches.

xuanyuanking · 2021-07-08T13:24:19Z

Thanks @HeartSaVioR and @viirya for all the help and review!
Finally, all the major implementations of the RocksDB state store have been merged!

SparkQA · 2021-07-08T13:46:43Z

Test build #140790 has finished for PR 33187 at commit b6a9d0a.

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

xuanyuanking changed the title ~~[SPARK-35988][SS] The implementation for RocksDBStateStoreProvider~~ [WIP][SPARK-35988][SS] The implementation for RocksDBStateStoreProvider Jul 2, 2021

github-actions bot added SQL STRUCTURED STREAMING labels Jul 2, 2021

HeartSaVioR reviewed Jul 5, 2021

View reviewed changes

xuanyuanking added 3 commits July 6, 2021 14:01

initial implementation

4335bc8

fix

f6312ad

address comment

9ddbf58

xuanyuanking force-pushed the SPARK-35988 branch from f59a9e9 to 9ddbf58 Compare July 6, 2021 06:14

xuanyuanking changed the title ~~[WIP][SPARK-35988][SS] The implementation for RocksDBStateStoreProvider~~ [SPARK-35988][SS] The implementation for RocksDBStateStoreProvider Jul 6, 2021

Fix the UT which should be considered as HDFSBackend only

a886567

fix

e63a380

HeartSaVioR reviewed Jul 8, 2021

View reviewed changes

address comment

1ac5835

HeartSaVioR approved these changes Jul 8, 2021

View reviewed changes

viirya reviewed Jul 8, 2021

View reviewed changes

viirya approved these changes Jul 8, 2021

View reviewed changes

address comment

b6a9d0a

HeartSaVioR approved these changes Jul 8, 2021

View reviewed changes

HeartSaVioR closed this in 0621e78 Jul 8, 2021

xuanyuanking deleted the SPARK-35988 branch July 8, 2021 13:23


		override def id: StateStoreId = RocksDBStateStoreProvider.this.stateStoreId

		override def version: Long = lastVersion

Conversation

xuanyuanking commented Jul 2, 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 Jul 2, 2021

Uh oh!

SparkQA commented Jul 2, 2021

Uh oh!

SparkQA commented Jul 2, 2021

Uh oh!

SparkQA commented Jul 2, 2021

Uh oh!

SparkQA commented Jul 2, 2021

Uh oh!

HeartSaVioR left a comment

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!

HeartSaVioR commented Jul 6, 2021

Uh oh!

HeartSaVioR commented Jul 6, 2021

Uh oh!

xuanyuanking commented Jul 6, 2021

Uh oh!

SparkQA commented Jul 6, 2021

Uh oh!

SparkQA commented Jul 6, 2021

Uh oh!

SparkQA commented Jul 6, 2021

Uh oh!

SparkQA commented Jul 6, 2021

Uh oh!

SparkQA commented Jul 6, 2021

Uh oh!

SparkQA commented Jul 6, 2021

Uh oh!

SparkQA commented Jul 6, 2021

Uh oh!

SparkQA commented Jul 6, 2021

Uh oh!

HeartSaVioR left a comment

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!

xuanyuanking commented Jul 8, 2021

Uh oh!

HeartSaVioR left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 8, 2021

Uh oh!