You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2023/06/06 05:08:13 UTC
[spark] branch branch-3.4 updated: [SPARK-43973][SS][UI] Structured Streaming UI should display failed queries correctly
This is an automated email from the ASF dual-hosted git repository.
gengliang pushed a commit to branch branch-3.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.4 by this push:
new b70510695a6 [SPARK-43973][SS][UI] Structured Streaming UI should display failed queries correctly
b70510695a6 is described below
commit b70510695a619bb42fb9a74c6c8e85907fbdb339
Author: Kris Mok <kr...@databricks.com>
AuthorDate: Mon Jun 5 22:07:39 2023 -0700
[SPARK-43973][SS][UI] Structured Streaming UI should display failed queries correctly
### What changes were proposed in this pull request?
Handle the `exception` message from Structured Streaming's `QueryTerminatedEvent` in `StreamingQueryStatusListener`, so that the Structured Streaming UI can display the status and error message correctly for failed queries.
### Why are the changes needed?
The original implementation of the Structured Streaming UI had a copy-and-paste bug where it forgot to handle the `exception` from `QueryTerminatedEvent` in `StreamingQueryStatusListener.onQueryTerminated`, so that field is never updated in the UI data, i.e. it's always `None`. In turn, the UI would always show the status of failed queries as `FINISHED` instead of `FAILED`.
### Does this PR introduce _any_ user-facing change?
Yes. Failed Structured Streaming queries would show incorrectly as `FINISHED` before the fix:
and show correctly as `FAILED` after the fix:
The example query is:
```scala
implicit val ctx = spark.sqlContext
import org.apache.spark.sql.execution.streaming.MemoryStream
spark.conf.set("spark.sql.ansi.enabled", "true")
val inputData = MemoryStream[(Int, Int)]
val df = inputData.toDF().selectExpr("_1 / _2 as a")
inputData.addData((1, 2), (3, 4), (5, 6), (7, 0))
val testQuery = df.writeStream.format("memory").queryName("kristest").outputMode("append").start
testQuery.processAllAvailable()
```
### How was this patch tested?
Added UT to `StreamingQueryStatusListenerSuite`
Closes #41468 from rednaxelafx/fix-streaming-ui.
Authored-by: Kris Mok <kr...@databricks.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
(cherry picked from commit 51a919ea8d65e38f829717ae822f31a1a7f57beb)
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../ui/StreamingQueryStatusListener.scala | 2 +-
.../ui/StreamingQueryStatusListenerSuite.scala | 34 ++++++++++++++++++++++
2 files changed, 35 insertions(+), 1 deletion(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala
index c5ecdb6395a..231ff07d91e 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListener.scala
@@ -107,7 +107,7 @@ private[sql] class StreamingQueryStatusListener(
querySummary.id,
querySummary.runId,
isActive = false,
- querySummary.exception,
+ event.exception,
querySummary.startTimestamp,
Some(curTime)
), checkTriggers = true)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListenerSuite.scala
index 58e04eb285a..4b6d2dc4562 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListenerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/ui/StreamingQueryStatusListenerSuite.scala
@@ -267,6 +267,40 @@ class StreamingQueryStatusListenerSuite extends StreamTest {
store.write(testData)
store.close(closeParent = false)
}
+
+ test("SPARK-43973: onQueryTerminated should pick up exception info") {
+ val kvStore = new ElementTrackingStore(createStore(), sparkConf)
+ val listener = new StreamingQueryStatusListener(spark.sparkContext.conf, kvStore)
+ val queryStore = new StreamingQueryStatusStore(kvStore)
+
+ // succeed (no exception) case
+ val id1 = UUID.randomUUID()
+ val runId1 = UUID.randomUUID()
+ val startEvent1 = new StreamingQueryListener.QueryStartedEvent(
+ id1, runId1, "test1", "2023-01-01T20:50:00.800Z")
+ listener.onQueryStarted(startEvent1)
+ val terminateEvent1 = new StreamingQueryListener.QueryTerminatedEvent(id1, runId1, None, None)
+ listener.onQueryTerminated(terminateEvent1)
+
+ // failure (has exception) case
+ val id2 = UUID.randomUUID()
+ val runId2 = UUID.randomUUID()
+ val startEvent2 = new StreamingQueryListener.QueryStartedEvent(
+ id2, runId2, "test2", "2023-01-02T20:54:20.827Z")
+ listener.onQueryStarted(startEvent2)
+ val terminateEvent2 = new StreamingQueryListener.QueryTerminatedEvent(
+ id2, runId2, Option("ExampleException"), Option("EXAMPLE_ERROR_CLASS"))
+ listener.onQueryTerminated(terminateEvent2)
+
+ // check results
+ val (activeQueries, stoppedQueries) = queryStore.allQueryUIData.partition(_.summary.isActive)
+ assert(activeQueries.isEmpty)
+ val (finishedQueries, failedQueries) = stoppedQueries.partition(_.summary.exception.isEmpty)
+ assert(finishedQueries.size == 1)
+ assert(failedQueries.size == 1)
+ assert(failedQueries.head.summary.exception == Option("ExampleException"))
+ // there's no UI state for errorClassOnException yet; should check it as well when it's added
+ }
}
class StreamingQueryStatusListenerWithDiskStoreSuite extends StreamingQueryStatusListenerSuite {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org