[SPARK-30915][SS] CompactibleFileStreamLog: Avoid reading the metadata log file when finding the latest batch ID #27664

HeartSaVioR · 2020-02-21T08:24:00Z

What changes were proposed in this pull request?

This patch adds the new method getLatestBatchId() in CompactibleFileStreamLog in complement of getLatest() which doesn't read the content of the latest batch metadata log file, and apply to both FileStreamSource and FileStreamSink to avoid unnecessary latency on reading log file.

Why are the changes needed?

Once compacted metadata log file becomes huge, writing outputs for the compact + 1 batch is also affected due to unnecessarily reading the compacted metadata log file. This unnecessary latency can be simply avoided.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

New UT. Also manually tested under query which has huge metadata log on file stream sink:

before applying the patch

after applying the patch

Peaks are compact batches - please compare the next batch after compact batches, especially the area of "light brown".

…ile when finding the latest batch ID

HeartSaVioR · 2020-02-21T08:27:32Z

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

+    val batchIds = fileManager.list(metadataPath, batchFilesFilter)
+      .map(f => pathToBatchId(f.getPath))
+      .sorted(Ordering.Long.reverse)
+    for (batchId <- batchIds) {


I just simply remove reading file here, but as we already get batch IDs from "listing" files, it may not even need to check for existence. It won't be the outstanding latency, though.

HeartSaVioR · 2020-02-21T08:29:12Z

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

    log.allFiles()
  }
+
+  private def withCountOpenLocalFileSystemAsLocalFileSystem(body: => Unit): Unit = {


The code regarding FileSystem I add here is very similar with what I add in #27620. When either one gets merged, I'll rebase and deduplicate it.

HeartSaVioR · 2020-02-21T08:30:49Z

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

    ))
  }

+  test("getLatestBatchId") {


I don't add E2E test to simplify the test code, but if we prefer E2E than I'll try to add a new test to FileStreamSinkSuite.

HeartSaVioR · 2020-02-21T08:45:26Z

cc. @tdas @zsxwing @gaborgsomogyi

SparkQA · 2020-02-21T13:33:40Z

Test build #118766 has finished for PR 27664 at commit 0cd8cda.

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

gaborgsomogyi

I've called both functions in the unit test 10k times and the difference looks good:

17:17:19.436 ERROR org.apache.spark.sql.execution.streaming.FileStreamSinkLogSuite: getLatestBatchId: 1449956803 ns
17:17:22.074 ERROR org.apache.spark.sql.execution.streaming.FileStreamSinkLogSuite: getLatest: 2638354455 ns

gaborgsomogyi · 2020-04-02T14:34:51Z

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

+          def getCountForOpenOnMetadataFile(batchId: Long): Long = {
+            val path = sinkLog.batchIdToPath(batchId).toUri.getPath
+            CountOpenLocalFileSystem.pathToNumOpenCalled
+              .get(path).map(_.get()).getOrElse(0)


Nit: no linebreak needed.

gaborgsomogyi · 2020-04-02T14:35:20Z

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

+    withCountOpenLocalFileSystemAsLocalFileSystem {
+      val scheme = CountOpenLocalFileSystem.scheme
+      withSQLConf(SQLConf.FILE_SINK_LOG_COMPACT_INTERVAL.key -> "3") {
+        withTempDir { file =>


If it's a dir maybe we can call it dir or path.

gaborgsomogyi · 2020-04-02T14:44:10Z

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

+
+          val curCount = getCountForOpenOnMetadataFile(2)
+
+          assert(sinkLog.getLatestBatchId() === Some(2))


Nit: s/2/2L/

gaborgsomogyi · 2020-04-02T14:44:37Z

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

+              .get(path).map(_.get()).getOrElse(0)
+          }
+
+          val curCount = getCountForOpenOnMetadataFile(2)


Nit: s/2/2L/

gaborgsomogyi · 2020-04-02T15:05:09Z

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

+
+object CountOpenLocalFileSystem {
+  val scheme = s"FileStreamSinkLogSuite${math.abs(Random.nextInt)}fs"
+  val pathToNumOpenCalled = new mutable.HashMap[String, AtomicLong]


Some reset functionality would be good to make it re-usable. This would also make curCount disappear.

Nit: could you use a thread-safe map in case someone may reuse this in a streaming query?

Yeah, I feel it's safer to make the map be thread-safe. I thought it's a contradiction to assume because we allow "resetting" the count, but it can be handled with caution.

gaborgsomogyi · 2020-04-02T15:07:24Z

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

+
+          assert(sinkLog.getLatestBatchId() === Some(2))
+          // getLatestBatchId doesn't open the latest metadata log file
+          assert(getCountForOpenOnMetadataFile(2L) === curCount)


Maybe worth to check other batches as well.

HeartSaVioR · 2020-04-03T03:36:18Z

Thanks for reviewing! Reflected review comments.

SparkQA · 2020-04-03T07:05:02Z

Test build #120748 has finished for PR 27664 at commit d270961.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

gaborgsomogyi · 2020-04-03T07:42:12Z

Seems unrelated.

gaborgsomogyi · 2020-04-03T07:42:19Z

retest this please

gaborgsomogyi

LGTM (pending tests)

gaborgsomogyi · 2020-04-03T07:54:12Z

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

+          assert(sinkLog.getLatestBatchId() === Some(2L))
+          // getLatestBatchId doesn't open the latest metadata log file
+          (0L to 2L).foreach { batchId =>
+            assert(getCountForOpenOnMetadataFile(batchId) === 0)


Nit: Just to be consistent with the other parts s/0/0L. Same applies to the other 2 occurrence...

Just replaced all constants where the type is Long. Thanks!

SparkQA · 2020-04-03T12:32:56Z

Test build #120764 has finished for PR 27664 at commit d270961.

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

SparkQA · 2020-04-03T16:14:19Z

Test build #120772 has finished for PR 27664 at commit 30338fb.

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

HeartSaVioR · 2020-04-13T22:52:11Z

retest this, please

SparkQA · 2020-04-14T03:46:45Z

Test build #121229 has finished for PR 27664 at commit 30338fb.

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

HeartSaVioR · 2020-04-15T06:54:24Z

retest this, please

SparkQA · 2020-04-15T12:08:55Z

Test build #121301 has finished for PR 27664 at commit 30338fb.

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

xuanyuanking · 2020-04-17T13:12:13Z

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

  }

  override def addBatch(batchId: Long, data: DataFrame): Unit = {
-    if (batchId <= fileLog.getLatest().map(_._1).getOrElse(-1L)) {


Nice catch!

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

Line 196 in 7ad6ba3

var latestId = getLatest().map(_._1).getOrElse(-1L)

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

Line 99 in 7ad6ba3

val latestBatchId = getLatest().map(_._1).getOrElse(-1L)

Can these two places also be optimized in this way?

Yeah I think so. Nice finding. Thanks!

SparkQA · 2020-04-17T16:08:59Z

Test build #121418 has finished for PR 27664 at commit ff9078e.

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

HeartSaVioR · 2020-04-18T00:06:38Z

retest this, please

SparkQA · 2020-04-18T05:39:55Z

Test build #121434 has finished for PR 27664 at commit ff9078e.

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

xuanyuanking

LGTM
Just one more nit: since the new function shares most of the code with the original getLatest, how about modifying getLatest with a default false param to control the file reading behavior?

HeartSaVioR · 2020-04-27T08:29:01Z

I'm sorry, but I think the name of the method should represent the thing what it does. That's why the boolean parameters have been considered as code smell and even we encourage to specify the name of boolean parameter while calling the method.

We may be able to try deduplicating the code via allowing some additional call on exists - so getLatest() can be implemented as a combination of getLatestBatchId() and get(). Does it make sense?

xuanyuanking · 2020-04-28T07:19:45Z

so getLatest() can be implemented as a combination of getLatestBatchId() and get(). Does it make sense?

Yeah, that would be good enough. No more duplicated code and the two similar functions will be in the same class.

SparkQA · 2020-04-29T07:05:01Z

Test build #122027 has finished for PR 27664 at commit f6078bb.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-04-29T07:05:02Z

Test build #122026 has finished for PR 27664 at commit fc00a0c.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

HeartSaVioR · 2020-04-29T07:35:35Z

retest this, please

SparkQA · 2020-04-29T11:59:26Z

Test build #122046 has finished for PR 27664 at commit f6078bb.

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

HeartSaVioR · 2020-04-29T12:04:57Z

retest this, please

SparkQA · 2020-04-29T16:43:02Z

Test build #122060 has finished for PR 27664 at commit f6078bb.

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

xuanyuanking

LGTM except for the new exception.
cc @zsxwing @brkyvz @tdas

xuanyuanking · 2020-05-13T06:08:38Z

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

+      val content = get(batchId).getOrElse {
+        // This only happens in odd case where the file exists when getLatestBatchId() is called,
+        // but get() doesn't find it.
+        throw new IllegalStateException(s"failed to read log file for batch $batchId")


nit but maybe not to involve a new behavior change here?

This is a part of the change in #25965 which should be dealt with. It shouldn't give the content with batch ID which is less than the latest batch ID - it should rather fail.

That said, you might have interesting proposals on my old PRs, https://github.com/apache/spark/pulls/HeartSaVioR

Thanks for the reference, will take a look later.

Thanks for reference #25965, LGTM to this change.
I personally think the comment in https://github.com/apache/spark/pull/25965/files#diff-aaeb546880508bb771df502318c40a99R183 is clearer. Either way is fine though.

Just pulled the comment here. Either this or #25965 will have to resolve merge conflict but wanted to be sure the code comment is clear in any way.

SparkQA · 2020-05-21T18:48:44Z

Test build #122933 has finished for PR 27664 at commit 83451c1.

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

zsxwing

Looks pretty good except some nits.

zsxwing · 2020-05-22T14:41:45Z

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

-      if (batch.isDefined) {
-        return Some((batchId, batch.get))
+      val batchMetadataFile = batchIdToPath(batchId)
+      if (fileManager.exists(batchMetadataFile)) {


Ni: I think we can remove this check. Previously, it was calling get which returns Option, so it had an extra isDefine check.

zsxwing · 2020-05-22T14:46:47Z

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

+        throw new IllegalStateException(s"failed to read log file for batch $batchId")
+      }
+      (batchId, content)
+    }.orElse(None)


nit: .orElse(None) is not needed.

zsxwing · 2020-05-22T14:49:00Z

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

+
+object CountOpenLocalFileSystem {
+  val scheme = s"FileStreamSinkLogSuite${math.abs(Random.nextInt)}fs"
+  val pathToNumOpenCalled = new mutable.HashMap[String, AtomicLong]


Nit: could you use a thread-safe map in case someone may reuse this in a streaming query?

zsxwing

LGTM pending tests

SparkQA · 2020-05-22T21:07:54Z

Test build #123009 has finished for PR 27664 at commit 4a5679e.

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

zsxwing · 2020-05-22T23:44:12Z

Thanks! Merging to master.

HeartSaVioR · 2020-05-23T00:17:25Z

Thanks all for reviewing and merging!

[SPARK-30915][SS] FileStreamSinkLog: Avoid reading the metadata log f…

0cd8cda

…ile when finding the latest batch ID

HeartSaVioR commented Feb 21, 2020

View reviewed changes

HeartSaVioR mentioned this pull request Feb 21, 2020

[SPARK-30866][SS] FileStreamSource: Cache fetched list of files beyond maxFilesPerTrigger as unread files #27620

Closed

HeartSaVioR commented Feb 21, 2020

View reviewed changes

dongjoon-hyun added the STRUCTURED STREAMING label Feb 28, 2020

gaborgsomogyi reviewed Apr 2, 2020

View reviewed changes

Reflect review comments

d270961

gaborgsomogyi approved these changes Apr 3, 2020

View reviewed changes

add 'L' where the constant should be Long

30338fb

xuanyuanking reviewed Apr 17, 2020

View reviewed changes

Address review comment (apply to more places)

ff9078e

probot-autolabeler bot added the SQL label Apr 17, 2020

HeartSaVioR changed the title ~~[SPARK-30915][SS] FileStreamSink: Avoid reading the metadata log file when finding the latest batch ID~~ [SPARK-30915][SS] CompactibleFileStreamLog: Avoid reading the metadata log file when finding the latest batch ID Apr 17, 2020

xuanyuanking reviewed Apr 22, 2020

View reviewed changes

Deduplicate code

f6078bb

HeartSaVioR force-pushed the SPARK-30915 branch from fc00a0c to f6078bb Compare April 29, 2020 02:49

xuanyuanking approved these changes May 13, 2020

View reviewed changes

Update comment to make clearer

83451c1

zsxwing reviewed May 22, 2020

View reviewed changes

Reflect review comments

4a5679e

zsxwing approved these changes May 22, 2020

View reviewed changes

asfgit closed this in 5a258b0 May 22, 2020

HeartSaVioR deleted the SPARK-30915 branch May 23, 2020 00:17


		val curCount = getCountForOpenOnMetadataFile(2)

		assert(sinkLog.getLatestBatchId() === Some(2))

[SPARK-30915][SS] CompactibleFileStreamLog: Avoid reading the metadata log file when finding the latest batch ID #27664

[SPARK-30915][SS] CompactibleFileStreamLog: Avoid reading the metadata log file when finding the latest batch ID #27664

Uh oh!

Conversation

HeartSaVioR commented Feb 21, 2020 • edited 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!

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 Feb 21, 2020

Uh oh!

SparkQA commented Feb 21, 2020

Uh oh!

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

HeartSaVioR commented Apr 3, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Apr 3, 2020

Uh oh!

gaborgsomogyi commented Apr 3, 2020

Uh oh!

gaborgsomogyi commented Apr 3, 2020

Uh oh!

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

SparkQA commented Apr 3, 2020

Uh oh!

SparkQA commented Apr 3, 2020

Uh oh!

HeartSaVioR commented Apr 13, 2020

Uh oh!

SparkQA commented Apr 14, 2020

Uh oh!

HeartSaVioR commented Apr 15, 2020

Uh oh!

SparkQA commented Apr 15, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 17, 2020

Uh oh!

HeartSaVioR commented Apr 18, 2020

Uh oh!

SparkQA commented Apr 18, 2020

Uh oh!

xuanyuanking left a comment

Choose a reason for hiding this comment

HeartSaVioR commented Feb 21, 2020 •

edited

Loading

HeartSaVioR commented Apr 3, 2020 •

edited

Loading