[SPARK-28784][SS] Use CheckpointFileManager in StreamExecution/StreamingQueryManager for checkpoint dirs #25514

shrutig · 2019-08-20T18:58:04Z

What changes were proposed in this pull request?

After SPARK-23966 Refactoring all checkpoint file writing logic in a common CheckpointFileManager interface, the CheckpointFileManager interface was created to handle all structured streaming checkpointing operations and helps users to choose how they wish to write checkpointing files atomically.
StreamExecution and StreamingQueryManager still uses some FileSystem operations without using the CheckpointFileManager.
For instance,

Instead, StreamExecution and StreamingQueryManager should use CheckpointFileManager for these operations.

Why are the changes needed?

This change will allow users to use CheckpointFileManager for structured streaming checkpointing files without need for a separate FileSystem implementation for the same.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing tests

…ntFileManager

shrutig · 2019-08-20T19:12:40Z

@tdas Could you please take a look at this PR ?

SparkQA · 2019-08-20T20:17:50Z

Test build #109428 has finished for PR 25514 at commit 97bcf5b.

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

dongjoon-hyun · 2019-08-21T19:01:12Z

Thank you for your first contribution, @shrutig .
Could you add a unit test case for your contribution? That will protect your contribution from the accidental future regressions.

dongjoon-hyun · 2019-08-21T19:01:19Z

Retest this please.

shrutig · 2019-08-21T22:02:16Z

Updated DataStreamReaderWriterSuite tests which were relying on resolvedCheckpointRoot variable in StreamExecution to use the checkpointFileManager method.
Plus, there is another test "should resolve the checkpoint path" already existing in StreamSuite file.

SparkQA · 2019-08-21T22:04:40Z

Test build #109519 has finished for PR 25514 at commit 97bcf5b.

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

…PARK-28784

SparkQA · 2019-08-22T00:36:57Z

Test build #109528 has finished for PR 25514 at commit 9a92b5e.

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

dongjoon-hyun

Please ignore this. (I removed my previous comment since it's outdated.)

zsxwing · 2019-08-22T22:16:11Z

IMO, it's not worth to switch to CheckpointFileManager considering we just found some issues in FileContext's checksum. See https://issues.apache.org/jira/browse/SPARK-28025. I'm inclined to use mature FileSystem APIs for now.

… tests to reflect changes make for CheckpointFileManager usages

shrutig · 2019-08-26T23:55:21Z

@dongjoon-hyun I was able to replicate the failing test in my local and I have submitted my changes to test files to fix that.

shrutig · 2019-08-27T00:10:15Z

@zsxwing CheckpointFileManager is already being used to handle checkpoint files in multiple places, such as HDFSMetadataLog, HDFSBackedStateStoreProvider and others as part of PR #21048 .
This PR aims to make that change uniformly in other classes.

As far as the issue in https://issues.apache.org/jira/browse/SPARK-28025 is concerned, in my opinion that should not hinder this PR. That issue has been approved and merged as part of a PR - #25565 . Also, it was caused due to a bug in a specific implementation of CheckpointFileManager (FileContextBasedCheckpointFileManager).

SparkQA · 2019-08-27T00:45:55Z

Test build #109753 has finished for PR 25514 at commit 73ff29b.

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

dongjoon-hyun · 2019-08-30T19:31:47Z

Retest this please.

dongjoon-hyun · 2019-08-30T19:35:48Z

Hi, @zsxwing .
If this PR makes these consistently, isn't it easier for us to change this together later without leaving something behind?

zsxwing · 2019-08-30T19:49:40Z

IMO, I don't feel comfortable to move to a known broken API when the existing codes work.

dongjoon-hyun · 2019-08-30T20:48:09Z

I got it. Thank you for the guide, @zsxwing .

In this case, @shrutig . We had better close this PR. I know your intention, but we cannot merge this according to the above advice. Could you close this?

holdenk · 2019-08-30T20:58:10Z

Isn't the issues @zsxwing refers to fixed in #25488 ? Also if this is still broken then maybe we should revert all uses of CheckPointManager? It seems that whichever we do we should do consistently.

SparkQA · 2019-08-30T23:01:50Z

Test build #109960 has finished for PR 25514 at commit 73ff29b.

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

gaborgsomogyi · 2019-09-02T08:59:29Z

The issue not yet fixed, it may happen in https://issues.apache.org/jira/browse/HADOOP-16255.

shrutig · 2019-09-05T19:22:37Z

I want to clear up some concerns which were mentioned here:

CheckpointFileManager is an API which was added in https://issues.apache.org/jira/browse/SPARK-23966 ([SPARK-23966][SS] Refactoring all checkpoint file writing logic in a common CheckpointFileManager interface #21048) to enable users to plug in their custom implementation for the createAtomic, delete, exists, etc functions (https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CheckpointFileManager.scala#L48) . This API is today already widely used in HDFSBackedStateStoreProvider for checkpoint operations.
The default implementation of the CheckpointFileManager api is FileContextCheckpointFileManager which has a bug in the createAtomic function, leading to file leaks. This JIRA elaborates the same : https://issues.apache.org/jira/browse/SPARK-28025 .
The change I am proposing in this PR does not add any usage of createAtomic method. I am aware of no bugs in the delete, mkdirs and exists functions of default CheckpointFileManager implementation. So, this PR should not be adding to the problem of leaving file leaks. @zsxwing @gaborgsomogyi Can you please elaborate on your concerns and mention any known issues with the delete, mkdirs and exists functions of default CheckpointFileManager implementation ?

cc @tdas

holdenk · 2019-09-14T02:29:03Z

It also looks like the default implementation issue has been fixed in https://issues.apache.org/jira/browse/HADOOP-16255

Since the FileContextCheckpointFileManager doesn't have any known issues with these code paths that are being added, unifying access to the same data is probably better, and from the original PR (cbb41a0) adding the CheckpointFileManager it sounds like the goal is to make this plug-gable which only works if it is used consistently inside of Spark (although @tdas can correct me on that point).

holdenk · 2019-09-19T18:34:49Z

Jenkins ok to test.

holdenk · 2019-09-19T18:35:13Z

@dongjoon-hyun it looks like you aren't asking for any specific change anymore, is that correct?

gaborgsomogyi · 2019-09-19T18:43:24Z

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

@@ -133,9 +134,8 @@ abstract class StreamExecution(
            .stripMargin)
      }
    }
-    val checkpointDir = checkpointPath.makeQualified(fs.getUri, fs.getWorkingDirectory)


makeQualified disappeared from here, right?

I think so. We're using the checkpointFileManager for this behavior now in the PR so we are not directly making FS operations. Looking inside of the default FileSystem impl we don't explicitly translate this to a fully-qualified path, do you think that would be an issue?

Since this ends-up in behavior change it may break existing queries. I think there must be a reason when we do such change.

gaborgsomogyi · 2019-09-19T18:44:51Z

sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala

@@ -444,7 +444,9 @@ class StreamSuite extends StreamTest {
    val inputData = MemoryStream[String]
    val df = inputData.toDS().map(_ + "foo").groupBy("value").agg(count("*"))
    // Test StreamingQuery.display
+    val dir1 = Utils.createTempDir().getCanonicalFile
    val q = df.writeStream.queryName("memory_explain").outputMode("complete").format("memory")


Why do we need this (and similar changes)? The original test is running fine with temp checkpoint.

SparkQA · 2019-09-19T22:51:58Z

Test build #111021 has finished for PR 25514 at commit 73ff29b.

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

holdenk · 2019-09-20T16:45:18Z

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

@@ -106,7 +107,7 @@ abstract class StreamExecution(
        new Path(new Path(checkpointPath.toUri.toString).toUri.toString).toUri.toString
      val legacyCheckpointDirExists =
        try {
-          fs.exists(new Path(legacyCheckpointDir))


Here would it make sense to keep the fs check as well since we want to catch the previously created check point directories and suggest folks migrate?

holdenk · 2019-09-20T16:51:01Z

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

@@ -133,9 +134,8 @@ abstract class StreamExecution(
            .stripMargin)
      }
    }
-    val checkpointDir = checkpointPath.makeQualified(fs.getUri, fs.getWorkingDirectory)


I think so. We're using the checkpointFileManager for this behavior now in the PR so we are not directly making FS operations. Looking inside of the default FileSystem impl we don't explicitly translate this to a fully-qualified path, do you think that would be an issue?

Remove the outdated review.

dongjoon-hyun · 2019-12-05T19:48:17Z

Retest this please.

SparkQA · 2019-12-05T23:43:48Z

Test build #114916 has finished for PR 25514 at commit 73ff29b.

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

github-actions · 2020-03-15T00:14:45Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

Updated StreamExecution and StreamingQueryManager to utilize Checkpoi…

97bcf5b

…ntFileManager

dongjoon-hyun added the STRUCTURED STREAMING label Aug 21, 2019

Updated DataStreamReaderWriterSuite tests

9a92b5e

Merge branch 'master' of github.pie.apple.com:pie/apache-spark into S…

626992e

…PARK-28784

dongjoon-hyun previously requested changes Aug 22, 2019

View reviewed changes

updated StreamSuite, StreamingQuerySuite, DataStreamReaderWriterSuite…

73ff29b

… tests to reflect changes make for CheckpointFileManager usages

dongjoon-hyun changed the title ~~[SPARK-28784][SS]StreamExecution and StreamingQueryManager should utilize CheckpointFileManager to interact with checkpoint directories~~ [SPARK-28784][SS] Use CheckpointFileManager in StreamExecution/StreamingQueryManager for checkpoint dirs Aug 30, 2019

gaborgsomogyi reviewed Sep 19, 2019

View reviewed changes

holdenk reviewed Sep 20, 2019

View reviewed changes

dongjoon-hyun self-requested a review December 5, 2019 19:41

github-actions bot added the Stale label Mar 15, 2020

github-actions bot closed this Mar 16, 2020

[SPARK-28784][SS] Use CheckpointFileManager in StreamExecution/StreamingQueryManager for checkpoint dirs #25514

[SPARK-28784][SS] Use CheckpointFileManager in StreamExecution/StreamingQueryManager for checkpoint dirs #25514

Uh oh!

Conversation

shrutig commented Aug 20, 2019 • edited by dongjoon-hyun Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

shrutig commented Aug 20, 2019

Uh oh!

SparkQA commented Aug 20, 2019

Uh oh!

dongjoon-hyun commented Aug 21, 2019

Uh oh!

dongjoon-hyun commented Aug 21, 2019

Uh oh!

shrutig commented Aug 21, 2019

Uh oh!

SparkQA commented Aug 21, 2019

Uh oh!

SparkQA commented Aug 22, 2019

Uh oh!

dongjoon-hyun left a comment • 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!

zsxwing commented Aug 22, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

shrutig commented Aug 26, 2019

Uh oh!

shrutig commented Aug 27, 2019

Uh oh!

SparkQA commented Aug 27, 2019

Uh oh!

dongjoon-hyun commented Aug 30, 2019

Uh oh!

dongjoon-hyun commented Aug 30, 2019

Uh oh!

zsxwing commented Aug 30, 2019

Uh oh!

dongjoon-hyun commented Aug 30, 2019

Uh oh!

holdenk commented Aug 30, 2019

Uh oh!

SparkQA commented Aug 30, 2019

Uh oh!

gaborgsomogyi commented Sep 2, 2019

Uh oh!

shrutig commented Sep 5, 2019

Uh oh!

holdenk commented Sep 14, 2019

Uh oh!

holdenk commented Sep 19, 2019

Uh oh!

holdenk commented Sep 19, 2019

Uh oh!

gaborgsomogyi Sep 19, 2019

Choose a reason for hiding this comment

Uh oh!

holdenk Sep 20, 2019

Choose a reason for hiding this comment

Uh oh!

gaborgsomogyi Sep 20, 2019

Choose a reason for hiding this comment

Uh oh!

gaborgsomogyi Sep 19, 2019

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 19, 2019

Uh oh!

holdenk Sep 20, 2019

Choose a reason for hiding this comment

Uh oh!

holdenk Sep 20, 2019

Choose a reason for hiding this comment

Uh oh!

shrutig commented Aug 20, 2019 •

edited by dongjoon-hyun

Loading

dongjoon-hyun left a comment •

edited

Loading

zsxwing commented Aug 22, 2019 •

edited

Loading