[SPARK-29314][SS] Don't overwrite the metric "updated" of state operator to 0 if empty batch is run #25987

HeartSaVioR · 2019-10-01T10:13:41Z

What changes were proposed in this pull request?

This patch fixes the behavior of ProgressReporter which always overwrite the value of "updated" of state operator to 0 if there's no new data. The behavior is correct only when we copy the state progress from "previous" executed plan, meaning no batch has been run. (Nonzero value of "updated" would be odd if batch didn't run, so it was correct.)

It was safe to assume no data is no batch, but SPARK-24156 enables empty data can run the batch if Spark needs to deal with watermark. After the patch, it only overwrites the value if both two conditions are met: 1) no data 2) no batch.

Why are the changes needed?

Currently Spark doesn't reflect correct metrics when empty batch is run and this patch fixes it.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Modified UT. Note that FlatMapGroupsWithState increases the value of "updated" when state rows are removed.
Also manually tested via below query (not a simple query to test with spark-shell, as you'll meet closure issue in spark-shell while playing with state func):

query

case class RunningCount(count: Long)

object TestFlatMapGroupsWithState {
  def main(args: Array[String]): Unit = {
    import org.apache.spark.sql.SparkSession

    val ss = SparkSession
      .builder()
      .appName("TestFlatMapGroupsWithState")
      .getOrCreate()

    ss.conf.set("spark.sql.shuffle.partitions", "5")

    import ss.implicits._

    val stateFunc = (key: String, values: Iterator[String], state: GroupState[RunningCount]) => {
      if (state.hasTimedOut) {
        // End users are not restricted to remove the state here - they can update the
        // state as well. For example, event time session window would have list of
        // sessions here and it cannot remove entire state.
        state.update(RunningCount(-1))
        Iterator((key, "-1"))
      } else {
        val count = state.getOption.map(_.count).getOrElse(0L) + values.size
        state.update(RunningCount(count))
        state.setTimeoutDuration("1 seconds")
        Iterator((key, count.toString))
      }
    }

    implicit val sqlContext = ss.sqlContext
    val inputData = MemoryStream[String]

    val result = inputData
      .toDF()
      .as[String]
      .groupByKey { v => v }
      .flatMapGroupsWithState(OutputMode.Append(), GroupStateTimeout.ProcessingTimeTimeout())(stateFunc)

    val query = result
      .writeStream
      .format("memory")
      .option("queryName", "test")
      .outputMode("append")
      .trigger(Trigger.ProcessingTime("5 second"))
      .start()

    Thread.sleep(1000)

    var chIdx: Long = 0

    while (true) {
      (chIdx to chIdx + 4).map { idx => inputData.addData(idx.toString) }
      chIdx += 5
      // intentionally sleep much more than trigger to enable "empty" batch
      Thread.sleep(10 * 1000)
    }
  }
}

before the patch (batch 3 which was an "empty" batch)

{
   "id":"de945a5c-882b-4dae-aa58-cb8261cbaf9e",
   "runId":"f1eb6d0d-3cd5-48b2-a03b-5e989b6c151b",
   "name":"test",
   "timestamp":"2019-11-18T07:00:25.005Z",
   "batchId":3,
   "numInputRows":0,
   "inputRowsPerSecond":0.0,
   "processedRowsPerSecond":0.0,
   "durationMs":{
      "addBatch":1664,
      "getBatch":0,
      "latestOffset":0,
      "queryPlanning":29,
      "triggerExecution":1789,
      "walCommit":51
   },
   "stateOperators":[
      {
         "numRowsTotal":10,
         "numRowsUpdated":0,
         "memoryUsedBytes":5130,
         "customMetrics":{
            "loadedMapCacheHitCount":15,
            "loadedMapCacheMissCount":0,
            "stateOnCurrentVersionSizeBytes":2722
         }
      }
   ],
   "sources":[
      {
         "description":"MemoryStream[value#1]",
         "startOffset":9,
         "endOffset":9,
         "numInputRows":0,
         "inputRowsPerSecond":0.0,
         "processedRowsPerSecond":0.0
      }
   ],
   "sink":{
      "description":"MemorySink",
      "numOutputRows":5
   }
}

after the patch (batch 3 which was an "empty" batch)

{
   "id":"7cb41623-6b9a-408e-ae02-6796ec636fa0",
   "runId":"17847710-ddfe-45f5-a7ab-b160e149382f",
   "name":"test",
   "timestamp":"2019-11-18T07:02:25.005Z",
   "batchId":3,
   "numInputRows":0,
   "inputRowsPerSecond":0.0,
   "processedRowsPerSecond":0.0,
   "durationMs":{
      "addBatch":1196,
      "getBatch":0,
      "latestOffset":0,
      "queryPlanning":30,
      "triggerExecution":1333,
      "walCommit":46
   },
   "stateOperators":[
      {
         "numRowsTotal":10,
         "numRowsUpdated":5,
         "memoryUsedBytes":5130,
         "customMetrics":{
            "loadedMapCacheHitCount":15,
            "loadedMapCacheMissCount":0,
            "stateOnCurrentVersionSizeBytes":2722
         }
      }
   ],
   "sources":[
      {
         "description":"MemoryStream[value#1]",
         "startOffset":9,
         "endOffset":9,
         "numInputRows":0,
         "inputRowsPerSecond":0.0,
         "processedRowsPerSecond":0.0
      }
   ],
   "sink":{
      "description":"MemorySink",
      "numOutputRows":5
   }
}

"numRowsUpdated" is 0 in "stateOperators" before applying the patch which is "wrong", as we "update" the state when timeout occurs. After applying the patch, it correctly represents the "numRowsUpdated" as 5 in "stateOperators".

HeartSaVioR · 2019-10-01T10:26:46Z

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

      },
      CheckNewAnswer(("c", "-1")),
-      assertNumStateRows(total = 0, updated = 0)
+      assertNumStateRows(total = 0, updated = 1)


This represents the patch is correct fix, though total = 0, updated = 1 might make someone feeling odd. This is because FlatMapGroupsWithState counts deletion of state as "update", whereas we don't count it as "update" in native streaming aggregation.

are there other tests where we can check this fix as well?

Actually the change we encounter the issue is small, as the bug is affecting the metric only when no data batch updates the state row. This is not occurred in non-arbitrary stateful operations, because possible state update is only eviction in this case, and state row eviction is not counted as updates.

This can be occurred in arbitrary stateful operations as timer can be triggered for no data batch and the query can update/delete state row which would trigger update count. This test is verifying one of these cases (delete state), hence I didn't add the new test. If we would like to have another test for state func to update the state row on timeout I can do, but most of things would be redundant.

SparkQA · 2019-10-01T13:44:20Z

Test build #111642 has finished for PR 25987 at commit 9866c9f.

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

HeartSaVioR · 2019-10-01T23:16:12Z

cc. @tdas @zsxwing @jose-torres @gaborgsomogyi

HeartSaVioR · 2019-11-18T07:16:14Z

I've added the query and changed outputs to show which issue this PR fixes.

HeartSaVioR · 2019-11-18T07:16:20Z

retest this, please

SparkQA · 2019-11-18T08:05:02Z

Test build #113997 has finished for PR 25987 at commit 9866c9f.

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

HeartSaVioR · 2019-11-18T08:12:44Z

Looks like known CI shutdown.

HeartSaVioR · 2019-11-18T08:12:51Z

retest this, please

SparkQA · 2019-11-18T12:00:36Z

Test build #114002 has finished for PR 25987 at commit 9866c9f.

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

SparkQA · 2019-12-06T16:56:50Z

Test build #114943 has finished for PR 25987 at commit 0f59ee2.

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

HeartSaVioR · 2020-01-13T02:58:24Z

retest this, please

SparkQA · 2020-01-13T07:23:20Z

Test build #116586 has finished for PR 25987 at commit 0f59ee2.

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

SparkQA · 2020-02-16T08:05:01Z

Test build #118492 has finished for PR 25987 at commit c6993d0.

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

HeartSaVioR · 2020-02-16T08:52:07Z

Retest this, please

SparkQA · 2020-02-16T12:49:41Z

Test build #118496 has finished for PR 25987 at commit c6993d0.

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

brkyvz

Are there other spots where we can add tests to @HeartSaVioR ?

brkyvz · 2020-04-08T00:21:57Z

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

you can just use runBatch here, or rather the new name hasExecuted

Yeah I'll resolve conflict and use hasExecuted. Thanks!

brkyvz · 2020-04-08T00:23:07Z

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

      },
      CheckNewAnswer(("c", "-1")),
-      assertNumStateRows(total = 0, updated = 0)
+      assertNumStateRows(total = 0, updated = 1)


are there other tests where we can check this fix as well?

…to 0 if empty batch is run

HeartSaVioR · 2020-04-08T06:58:02Z

Thanks for reviewing. Just rebased and reflected review comment.

Btw, I've left a comment to explain why I simply modified the existing test instead - #25987 (comment)

SparkQA · 2020-04-08T11:19:48Z

Test build #120953 has finished for PR 25987 at commit 983d79f.

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

brkyvz

LGTM

brkyvz · 2020-04-08T23:58:41Z

Merging to master and 3.0. Thanks @HeartSaVioR

…tor to 0 if empty batch is run ### What changes were proposed in this pull request? This patch fixes the behavior of ProgressReporter which always overwrite the value of "updated" of state operator to 0 if there's no new data. The behavior is correct only when we copy the state progress from "previous" executed plan, meaning no batch has been run. (Nonzero value of "updated" would be odd if batch didn't run, so it was correct.) It was safe to assume no data is no batch, but SPARK-24156 enables empty data can run the batch if Spark needs to deal with watermark. After the patch, it only overwrites the value if both two conditions are met: 1) no data 2) no batch. ### Why are the changes needed? Currently Spark doesn't reflect correct metrics when empty batch is run and this patch fixes it. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Modified UT. Note that FlatMapGroupsWithState increases the value of "updated" when state rows are removed. Also manually tested via below query (not a simple query to test with spark-shell, as you'll meet closure issue in spark-shell while playing with state func): > query ``` case class RunningCount(count: Long) object TestFlatMapGroupsWithState { def main(args: Array[String]): Unit = { import org.apache.spark.sql.SparkSession val ss = SparkSession .builder() .appName("TestFlatMapGroupsWithState") .getOrCreate() ss.conf.set("spark.sql.shuffle.partitions", "5") import ss.implicits._ val stateFunc = (key: String, values: Iterator[String], state: GroupState[RunningCount]) => { if (state.hasTimedOut) { // End users are not restricted to remove the state here - they can update the // state as well. For example, event time session window would have list of // sessions here and it cannot remove entire state. state.update(RunningCount(-1)) Iterator((key, "-1")) } else { val count = state.getOption.map(_.count).getOrElse(0L) + values.size state.update(RunningCount(count)) state.setTimeoutDuration("1 seconds") Iterator((key, count.toString)) } } implicit val sqlContext = ss.sqlContext val inputData = MemoryStream[String] val result = inputData .toDF() .as[String] .groupByKey { v => v } .flatMapGroupsWithState(OutputMode.Append(), GroupStateTimeout.ProcessingTimeTimeout())(stateFunc) val query = result .writeStream .format("memory") .option("queryName", "test") .outputMode("append") .trigger(Trigger.ProcessingTime("5 second")) .start() Thread.sleep(1000) var chIdx: Long = 0 while (true) { (chIdx to chIdx + 4).map { idx => inputData.addData(idx.toString) } chIdx += 5 // intentionally sleep much more than trigger to enable "empty" batch Thread.sleep(10 * 1000) } } } ``` > before the patch (batch 3 which was an "empty" batch) ``` { "id":"de945a5c-882b-4dae-aa58-cb8261cbaf9e", "runId":"f1eb6d0d-3cd5-48b2-a03b-5e989b6c151b", "name":"test", "timestamp":"2019-11-18T07:00:25.005Z", "batchId":3, "numInputRows":0, "inputRowsPerSecond":0.0, "processedRowsPerSecond":0.0, "durationMs":{ "addBatch":1664, "getBatch":0, "latestOffset":0, "queryPlanning":29, "triggerExecution":1789, "walCommit":51 }, "stateOperators":[ { "numRowsTotal":10, "numRowsUpdated":0, "memoryUsedBytes":5130, "customMetrics":{ "loadedMapCacheHitCount":15, "loadedMapCacheMissCount":0, "stateOnCurrentVersionSizeBytes":2722 } } ], "sources":[ { "description":"MemoryStream[value#1]", "startOffset":9, "endOffset":9, "numInputRows":0, "inputRowsPerSecond":0.0, "processedRowsPerSecond":0.0 } ], "sink":{ "description":"MemorySink", "numOutputRows":5 } } ``` > after the patch (batch 3 which was an "empty" batch) ``` { "id":"7cb41623-6b9a-408e-ae02-6796ec636fa0", "runId":"17847710-ddfe-45f5-a7ab-b160e149382f", "name":"test", "timestamp":"2019-11-18T07:02:25.005Z", "batchId":3, "numInputRows":0, "inputRowsPerSecond":0.0, "processedRowsPerSecond":0.0, "durationMs":{ "addBatch":1196, "getBatch":0, "latestOffset":0, "queryPlanning":30, "triggerExecution":1333, "walCommit":46 }, "stateOperators":[ { "numRowsTotal":10, "numRowsUpdated":5, "memoryUsedBytes":5130, "customMetrics":{ "loadedMapCacheHitCount":15, "loadedMapCacheMissCount":0, "stateOnCurrentVersionSizeBytes":2722 } } ], "sources":[ { "description":"MemoryStream[value#1]", "startOffset":9, "endOffset":9, "numInputRows":0, "inputRowsPerSecond":0.0, "processedRowsPerSecond":0.0 } ], "sink":{ "description":"MemorySink", "numOutputRows":5 } } ``` "numRowsUpdated" is `0` in "stateOperators" before applying the patch which is "wrong", as we "update" the state when timeout occurs. After applying the patch, it correctly represents the "numRowsUpdated" as `5` in "stateOperators". Closes #25987 from HeartSaVioR/SPARK-29314. Authored-by: Jungtaek Lim (HeartSaVioR) <[email protected]> Signed-off-by: Burak Yavuz <[email protected]> (cherry picked from commit ca2ba4f) Signed-off-by: Burak Yavuz <[email protected]>

HeartSaVioR · 2020-04-09T00:25:03Z

Thanks for reviewing and merging!

…tor to 0 if empty batch is run ### What changes were proposed in this pull request? This patch fixes the behavior of ProgressReporter which always overwrite the value of "updated" of state operator to 0 if there's no new data. The behavior is correct only when we copy the state progress from "previous" executed plan, meaning no batch has been run. (Nonzero value of "updated" would be odd if batch didn't run, so it was correct.) It was safe to assume no data is no batch, but SPARK-24156 enables empty data can run the batch if Spark needs to deal with watermark. After the patch, it only overwrites the value if both two conditions are met: 1) no data 2) no batch. ### Why are the changes needed? Currently Spark doesn't reflect correct metrics when empty batch is run and this patch fixes it. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Modified UT. Note that FlatMapGroupsWithState increases the value of "updated" when state rows are removed. Also manually tested via below query (not a simple query to test with spark-shell, as you'll meet closure issue in spark-shell while playing with state func): > query ``` case class RunningCount(count: Long) object TestFlatMapGroupsWithState { def main(args: Array[String]): Unit = { import org.apache.spark.sql.SparkSession val ss = SparkSession .builder() .appName("TestFlatMapGroupsWithState") .getOrCreate() ss.conf.set("spark.sql.shuffle.partitions", "5") import ss.implicits._ val stateFunc = (key: String, values: Iterator[String], state: GroupState[RunningCount]) => { if (state.hasTimedOut) { // End users are not restricted to remove the state here - they can update the // state as well. For example, event time session window would have list of // sessions here and it cannot remove entire state. state.update(RunningCount(-1)) Iterator((key, "-1")) } else { val count = state.getOption.map(_.count).getOrElse(0L) + values.size state.update(RunningCount(count)) state.setTimeoutDuration("1 seconds") Iterator((key, count.toString)) } } implicit val sqlContext = ss.sqlContext val inputData = MemoryStream[String] val result = inputData .toDF() .as[String] .groupByKey { v => v } .flatMapGroupsWithState(OutputMode.Append(), GroupStateTimeout.ProcessingTimeTimeout())(stateFunc) val query = result .writeStream .format("memory") .option("queryName", "test") .outputMode("append") .trigger(Trigger.ProcessingTime("5 second")) .start() Thread.sleep(1000) var chIdx: Long = 0 while (true) { (chIdx to chIdx + 4).map { idx => inputData.addData(idx.toString) } chIdx += 5 // intentionally sleep much more than trigger to enable "empty" batch Thread.sleep(10 * 1000) } } } ``` > before the patch (batch 3 which was an "empty" batch) ``` { "id":"de945a5c-882b-4dae-aa58-cb8261cbaf9e", "runId":"f1eb6d0d-3cd5-48b2-a03b-5e989b6c151b", "name":"test", "timestamp":"2019-11-18T07:00:25.005Z", "batchId":3, "numInputRows":0, "inputRowsPerSecond":0.0, "processedRowsPerSecond":0.0, "durationMs":{ "addBatch":1664, "getBatch":0, "latestOffset":0, "queryPlanning":29, "triggerExecution":1789, "walCommit":51 }, "stateOperators":[ { "numRowsTotal":10, "numRowsUpdated":0, "memoryUsedBytes":5130, "customMetrics":{ "loadedMapCacheHitCount":15, "loadedMapCacheMissCount":0, "stateOnCurrentVersionSizeBytes":2722 } } ], "sources":[ { "description":"MemoryStream[value#1]", "startOffset":9, "endOffset":9, "numInputRows":0, "inputRowsPerSecond":0.0, "processedRowsPerSecond":0.0 } ], "sink":{ "description":"MemorySink", "numOutputRows":5 } } ``` > after the patch (batch 3 which was an "empty" batch) ``` { "id":"7cb41623-6b9a-408e-ae02-6796ec636fa0", "runId":"17847710-ddfe-45f5-a7ab-b160e149382f", "name":"test", "timestamp":"2019-11-18T07:02:25.005Z", "batchId":3, "numInputRows":0, "inputRowsPerSecond":0.0, "processedRowsPerSecond":0.0, "durationMs":{ "addBatch":1196, "getBatch":0, "latestOffset":0, "queryPlanning":30, "triggerExecution":1333, "walCommit":46 }, "stateOperators":[ { "numRowsTotal":10, "numRowsUpdated":5, "memoryUsedBytes":5130, "customMetrics":{ "loadedMapCacheHitCount":15, "loadedMapCacheMissCount":0, "stateOnCurrentVersionSizeBytes":2722 } } ], "sources":[ { "description":"MemoryStream[value#1]", "startOffset":9, "endOffset":9, "numInputRows":0, "inputRowsPerSecond":0.0, "processedRowsPerSecond":0.0 } ], "sink":{ "description":"MemorySink", "numOutputRows":5 } } ``` "numRowsUpdated" is `0` in "stateOperators" before applying the patch which is "wrong", as we "update" the state when timeout occurs. After applying the patch, it correctly represents the "numRowsUpdated" as `5` in "stateOperators". Closes apache#25987 from HeartSaVioR/SPARK-29314. Authored-by: Jungtaek Lim (HeartSaVioR) <[email protected]> Signed-off-by: Burak Yavuz <[email protected]>

HeartSaVioR changed the title ~~[SPARK-29314][SS] Don't overwrite "updated" of state operator metric to 0 if empty batch is run~~ [SPARK-29314][SS] Don't overwrite the metric "updated" of state operator to 0 if empty batch is run Oct 1, 2019

HeartSaVioR mentioned this pull request Oct 1, 2019

[SPARK-24634][SS] Add a new metric regarding number of rows later than watermark plus allowed delay #24936

Closed

HeartSaVioR commented Oct 1, 2019

View reviewed changes

dongjoon-hyun added the STRUCTURED STREAMING label Oct 1, 2019

HeartSaVioR force-pushed the SPARK-29314 branch from 9866c9f to 0f59ee2 Compare December 6, 2019 12:58

HeartSaVioR force-pushed the SPARK-29314 branch from 0f59ee2 to c6993d0 Compare February 16, 2020 07:05

HeartSaVioR mentioned this pull request Mar 27, 2020

[SPARK-31278][SS] Fix StreamingQuery output rows metric #28040

Closed

brkyvz suggested changes Apr 8, 2020

View reviewed changes

[SPARK-29314][SS] Don't overwrite "updated" of state operator metric …

983d79f

…to 0 if empty batch is run

HeartSaVioR force-pushed the SPARK-29314 branch from c6993d0 to 983d79f Compare April 8, 2020 06:54

brkyvz approved these changes Apr 8, 2020

View reviewed changes

asfgit closed this in ca2ba4f Apr 9, 2020

HeartSaVioR deleted the SPARK-29314 branch April 13, 2020 23:43

[SPARK-29314][SS] Don't overwrite the metric "updated" of state operator to 0 if empty batch is run #25987

[SPARK-29314][SS] Don't overwrite the metric "updated" of state operator to 0 if empty batch is run #25987

Uh oh!

Conversation

HeartSaVioR commented Oct 1, 2019 • 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!

HeartSaVioR Oct 1, 2019 • 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!

brkyvz Apr 8, 2020

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Apr 8, 2020 • 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!

SparkQA commented Oct 1, 2019

Uh oh!

HeartSaVioR commented Oct 1, 2019

Uh oh!

HeartSaVioR commented Nov 18, 2019

Uh oh!

HeartSaVioR commented Nov 18, 2019

Uh oh!

SparkQA commented Nov 18, 2019

Uh oh!

HeartSaVioR commented Nov 18, 2019

Uh oh!

HeartSaVioR commented Nov 18, 2019

Uh oh!

SparkQA commented Nov 18, 2019

Uh oh!

SparkQA commented Dec 6, 2019

Uh oh!

HeartSaVioR commented Jan 13, 2020

Uh oh!

SparkQA commented Jan 13, 2020

Uh oh!

SparkQA commented Feb 16, 2020

Uh oh!

HeartSaVioR commented Feb 16, 2020

Uh oh!

SparkQA commented Feb 16, 2020

Uh oh!

brkyvz left a comment

Choose a reason for hiding this comment

Uh oh!

brkyvz Apr 8, 2020

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR Apr 8, 2020

Choose a reason for hiding this comment

Uh oh!

brkyvz Apr 8, 2020

Choose a reason for hiding this comment

Uh oh!

HeartSaVioR commented Apr 8, 2020

Uh oh!

SparkQA commented Apr 8, 2020

Uh oh!

brkyvz left a comment

Choose a reason for hiding this comment

Uh oh!

brkyvz commented Apr 8, 2020

Uh oh!

HeartSaVioR commented Apr 9, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

HeartSaVioR commented Oct 1, 2019 •

edited

Loading

HeartSaVioR Oct 1, 2019 •

edited

Loading

HeartSaVioR Apr 8, 2020 •

edited

Loading