[SPARK-35436][SS] RocksDBFileManager - save checkpoint to DFS #32582

xuanyuanking · 2021-05-18T14:55:12Z

What changes were proposed in this pull request?

The implementation for the save operation of RocksDBFileManager.

Why are the changes needed?

Save all the files in the given local checkpoint directory as a committed version in DFS.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

New UT added.

xuanyuanking · 2021-05-18T14:57:49Z

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

Per #32272 (comment), we only generate the RocksDBLogFile object in the operation of saving checkpoint here.

xuanyuanking · 2021-05-18T14:59:51Z

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

Per #32272 (comment), we only use prettyJson in the log.

SparkQA · 2021-05-18T15:41:27Z

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

dongjoon-hyun · 2021-05-18T15:50:32Z

cc @viirya

SparkQA · 2021-05-18T17:11:01Z

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

viirya · 2021-05-18T17:21:52Z

Thanks @dongjoon-hyun. I'll look at this.

SparkQA · 2021-05-18T17:44:37Z

Test build #138685 has finished for PR 32582 at commit 5e6bf13.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class RocksDBFileManager(

xuanyuanking · 2021-05-28T03:11:04Z

As we merged #32272, after rebasing and addressing the comment, this one is ready for review. cc @viirya and @HeartSaVioR

viirya · 2021-05-28T03:23:38Z

Thanks @xuanyuanking. I will find some time to review this.

SparkQA · 2021-05-28T04:27:36Z

Kubernetes integration test unable to build dist.

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

SparkQA · 2021-05-28T06:08:59Z

Test build #139041 has finished for PR 32582 at commit 703f59f.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class RocksDBFileManager(

HeartSaVioR

First pass.

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

HeartSaVioR · 2021-05-30T05:50:45Z

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

Just to confirm, could I safely assume versionToRocksDBFiles will be loaded when RocksDBFileManager is initialized with existing checkpoint in further PR?

That's right. versionToRocksDBFiles was touched in the following 3 places:

saveCheckpointToDfs (this PR)

deleteOldVersions

loadCheckpointFromDfs

HeartSaVioR · 2021-05-30T06:54:03Z

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

I'm confusing here; if I understand correctly, we use version as batch ID + 1 for HDFSBackedStateStoreProvider, so version 0 with numKeys > 0 doesn't make sense at least for now. (As we don't have a functionality of "initial state" for now.)

How it works for RocksDB state store provider? Would the version be same with batch ID?

We also use batch ID + 1 as the version for the RocksDB state store. Since the batch ID starts from -1 so for the corner case of the first empty batch (batch ID 0), we add the safeguard here to make sure the working dir is created.

What about simply making sure the directory has been created in prior? We can add below in initialization and don't bother with handling specific case.

// create a root directory on initialization if necessary if (!fm.exists(new Path(dfsRootDir))) { fm.mkdirs(new Path(dfsRootDir)) }

If we'd like to avoid paying the cost on calling exist, let's keep the code and change the version to 1 as I met the case the version starts from 1 during test.

Yeah, let's change the version to 1 for avoiding the extra cost of file operations. Will change it in the next commit. Done in bdd9e8e

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

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

HeartSaVioR · 2021-05-30T07:36:40Z

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

With this, def listFiles(file: String) is probably unnecessary, as I expect implicit conversion happening when calling listFiles(file: File) with String.

That's right in this PR. In the further PRs, we will have more caller for the listFiles(file: String). Maybe let's keep it for now.

HeartSaVioR · 2021-05-30T07:48:33Z

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

Shall we have some code to check other files & metadata as well? We don't seem to cover the test on existence of zip file and the content.

We'll enhance this test in the next PR, RocksDBFileManager - load checkpoint from DFS, by load back the metadata and check the other files. Let me keep this conversation open and be referenced in the next PR.

Referenced this comment to https://github.com/apache/spark/pull/32767/files#diff-dc6f9dfe11e76f890ff2986f866853bcac263027c82562f9a52f4672a5460826R79

HeartSaVioR · 2021-05-30T07:57:29Z

And I see lots of information regarding files will be printed out via INFO level. I don't know whether the information would make the log file being flooded, but I guess we want to leave the information for the first phase on ease of debugging. We could discuss to lower down the level if we are getting bothered from the log size.

xuanyuanking · 2021-06-02T11:10:24Z

Great thanks for the detailed review! All comments addressed. Please take another look when you have time.
Regarding the logs, agree on more discussion about lower down some flooded messages. I'll also revisit other logs we add. For this PR, 7 INFO logs are for recoding file operations and metrics.

SparkQA · 2021-06-02T11:50:25Z

Kubernetes integration test unable to build dist.

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

SparkQA · 2021-06-02T13:21:52Z

Test build #139221 has finished for PR 32582 at commit 3800c51.

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

HeartSaVioR

Thanks for the updates! The new changes look OK.

It would be nice if we can do the same we did for reviewing the first PR; if remaining review comments are planned to be addressed via further PR, please raise the PR so that we can refer the PR and finally approve this one.

xuanyuanking · 2021-06-03T03:32:01Z

It would be nice if we can do the same we did for reviewing the first PR; if remaining review comments are planned to be addressed via further PR, please raise the PR so that we can refer the PR and finally approve this one.

Agree, let me submit the 3rd PR today.

The failure tests are not related, they can pass locally.

viirya · 2021-06-03T03:48:35Z

Sorry for late. I will take look this soon.

viirya

Went through RocksDBFileManager, but not look at tests yet.

core/src/main/scala/org/apache/spark/util/Utils.scala

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

viirya · 2021-06-03T04:21:31Z

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

The zipped file is dfsZipFile, right? why filesStr here?

the fileStr contains the dfsZipFile name. The log format here is ${dfsZipFile} \n ${listing all the file names}

viirya · 2021-06-03T04:22:38Z

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

Can you talk more about there this claim (thread-safe) applies? Where we delete old files?

Do you mean versionToRocksDBFiles? When we prepare files for new version, there is another thread (maintenance thread) deleting old files?

Yes. It refers to delete the files contained in the old versions. Here's the description of the deleteOldVersions method of RocksDBFileManager, which will be called in RocksDBStateStoreProvider.doMaintenance. As we did before, I'll also refer this comment when the PR for delete path submitted.

* Delete old versions by deleting the associated version and SST files. * At a high-level, this method finds which versions to delete, and which SST files that were * last used in those versions. Its safe to delete these SST files because a SST file can * be reused only in successive versions. Therefore, if a SST file F was last used in version * V, then it wont be used in version V+1 or later, and if version V can be deleted, then * F can safely be deleted as well.

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

SparkQA · 2021-06-03T11:50:23Z

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

SparkQA · 2021-06-03T12:22:03Z

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

viirya · 2021-06-04T06:36:55Z

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

version = 1? In practice, will we save checkpoint files for same version? I.e. for same micro-batch?

For task failover, it's possible that the tasks failed during uploading. The new task uses the same version and may schedule to the same executor but with different checkpoint dir.

viirya · 2021-06-04T06:37:46Z

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

Use withTempDir too?

This test will be further extended and more temp dir will be added. So maybe we just have the root path using withTempDir? See the demo here: https://github.com/apache/spark/pull/32767/files#diff-dc6f9dfe11e76f890ff2986f866853bcac263027c82562f9a52f4672a5460826R37

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

viirya · 2021-06-04T06:40:51Z

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

I'm not sure why this will fail?

This is failed on purpose.

[info] java.io.IOException: Copy failed intentionally [info] at org.apache.spark.sql.execution.streaming.CreateAtomicTestManager$$anon$3.close(CheckpointFileManagerSuite.scala:169)

This test aims to test the behavior that the cancel should be called no matter any error happens during close. See the comment and code here: https://github.com/apache/spark/pull/32582/files#diff-e3d3914d0398d61fdd299b1f8d3e869ec6a86e97606677c724969e421c9bf44eR222-R227

SparkQA · 2021-06-04T16:53:31Z

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

SparkQA · 2021-06-04T17:27:48Z

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

SparkQA · 2021-06-04T18:35:11Z

Test build #139340 has finished for PR 32582 at commit bdd9e8e.

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

xuanyuanking · 2021-06-08T15:44:10Z

Gentle ping @viirya @HeartSaVioR Do you think this is ready to go?

viirya · 2021-06-08T23:06:17Z

Thanks @xuanyuanking for the work. Let's move forward to next related PR.

HeartSaVioR · 2021-06-09T01:05:40Z

I'll merge once either build test passes.

HeartSaVioR · 2021-06-09T01:05:45Z

retest this, please

xuanyuanking · 2021-06-09T01:05:56Z

Thanks @viirya. Let me retrigger the test.

SparkQA · 2021-06-09T02:24:56Z

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

SparkQA · 2021-06-09T03:10:57Z

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

SparkQA · 2021-06-09T04:13:21Z

Test build #139516 has finished for PR 32582 at commit f86cea4.

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

HeartSaVioR · 2021-06-09T05:07:59Z

Thanks! Merging to master.

HeartSaVioR · 2021-06-09T05:10:18Z

Thanks @xuanyuanking for your contribution! I merged this to master branch.

SparkQA · 2021-06-09T05:21:34Z

Kubernetes integration test unable to build dist.

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

SparkQA · 2021-06-09T06:13:17Z

Test build #139529 has finished for PR 32582 at commit c05b140.

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

xuanyuanking · 2021-06-09T06:26:20Z

Great thanks for the help! @HeartSaVioR

xuanyuanking commented May 18, 2021

View reviewed changes

xuanyuanking mentioned this pull request May 18, 2021

[SPARK-35172][SS] The implementation of RocksDBCheckpointMetadata #32272

Closed

github-actions bot added CORE SQL STRUCTURED STREAMING labels May 18, 2021

xuanyuanking force-pushed the SPARK-35436 branch from 5e6bf13 to 703f59f Compare May 28, 2021 03:06

xuanyuanking changed the title ~~[WIP][SPARK-35436] RocksDBFileManager - save checkpoint to DFS~~ [SPARK-35436] RocksDBFileManager - save checkpoint to DFS May 28, 2021

HeartSaVioR changed the title ~~[SPARK-35436] RocksDBFileManager - save checkpoint to DFS~~ [SPARK-35436][SS] RocksDBFileManager - save checkpoint to DFS May 28, 2021

HeartSaVioR reviewed May 30, 2021

View reviewed changes

HeartSaVioR reviewed Jun 3, 2021

View reviewed changes

viirya reviewed Jun 3, 2021

View reviewed changes

xuanyuanking mentioned this pull request Jun 3, 2021

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

Closed

viirya reviewed Jun 4, 2021

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/RocksDBSuite.scala Outdated Show resolved Hide resolved

viirya reviewed Jun 4, 2021

View reviewed changes

viirya approved these changes Jun 8, 2021

View reviewed changes

xuanyuanking added 4 commits June 9, 2021 11:31

implementation for RocksDBFileManager - save checkpoint to DFS

4d5b888

address comments

aae6415

address comment

72962ce

address comments

c05b140

xuanyuanking force-pushed the SPARK-35436 branch from f86cea4 to c05b140 Compare June 9, 2021 03:31

HeartSaVioR closed this in 9f010a8 Jun 9, 2021

xuanyuanking deleted the SPARK-35436 branch June 9, 2021 06:26

[SPARK-35436][SS] RocksDBFileManager - save checkpoint to DFS #32582

[SPARK-35436][SS] RocksDBFileManager - save checkpoint to DFS #32582

Uh oh!

Conversation

xuanyuanking commented May 18, 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented May 18, 2021

Uh oh!

dongjoon-hyun commented May 18, 2021

Uh oh!

SparkQA commented May 18, 2021

Uh oh!

viirya commented May 18, 2021

Uh oh!

SparkQA commented May 18, 2021

Uh oh!

xuanyuanking commented May 28, 2021

Uh oh!

viirya commented May 28, 2021

Uh oh!

SparkQA commented May 28, 2021

Uh oh!

SparkQA commented May 28, 2021

Uh oh!

HeartSaVioR left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

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!

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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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!

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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented May 30, 2021

Uh oh!

xuanyuanking commented Jun 2, 2021

Uh oh!

SparkQA commented Jun 2, 2021

Uh oh!

SparkQA commented Jun 2, 2021

Uh oh!

HeartSaVioR left a comment

Choose a reason for hiding this comment

Uh oh!

xuanyuanking commented Jun 3, 2021

Uh oh!

xuanyuanking Jun 4, 2021 •

edited

Loading

xuanyuanking Jun 3, 2021 •

edited

Loading