[SPARK-38564][SS] Support collecting metrics from streaming sinks #35872

jerrypeng · 2022-03-16T05:56:19Z

What changes were proposed in this pull request?

Add the capability for streaming sinks to report custom metrics just like streaming sources

Why are the changes needed?

Allowing streaming sinks to report custom metrics is useful and achieve feature parity with streaming sources

Does this PR introduce any user-facing change?

no

How was this patch tested?

New UT

HyukjinKwon · 2022-03-17T02:55:17Z

cc @HeartSaVioR FYI

AmplabJenkins · 2022-03-17T17:38:05Z

Can one of the admins verify this patch?

HeartSaVioR

The code change looks OK given the proposed interface is symmetric with source one.

Shall we add some tests for this new feature? We would like to be very sure about the functional behavior.

Thanks in advance!

HeartSaVioR · 2022-03-18T09:37:04Z

...catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/ReportsSinkMetrics.java

+ * A mix-in interface for streaming sinks to signal that they can report
+ * metrics.
+ *
+ * @since 3.3.0


nit: 3.4.0 as we missed to catch the train.

It seems to be missed. Could you please update this?

actually updated this but forgot to include in commit :(

jerrypeng · 2022-03-20T08:41:52Z

@HeartSaVioR thanks for the review! Please take another look!

HeartSaVioR · 2022-03-20T18:39:40Z

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

+    }
+
+    def createRelation(
+                        sqlContext: SQLContext,


nit: indent

HeartSaVioR · 2022-03-20T18:55:33Z

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

+
+      inputData.addData(1, 2, 3)
+
+      var metricsMap: java.util.Map[String, String] = null


It would be safer to register the listener before executing the query. Since you've started the query also added the data here, the execution of streaming query and registration of listener go concurrently.

I'd move the registration of listener out before try statement, and remove the listener in finally statement as we do for StreamingListenerQuerySuite, as a zen of defensive programming.

HeartSaVioR · 2022-03-20T18:58:15Z

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

+        df.writeStream
+          .outputMode("append")
+          .format("org.apache.spark.sql.streaming.TestSinkProvider")
+          .option("checkPointLocation", Files.createTempDirectory("some-prefix").toFile.getName)


nit: let's use withTempDir as following the practice on test code.

HeartSaVioR · 2022-03-20T19:00:07Z

...catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/ReportsSinkMetrics.java

+ * A mix-in interface for streaming sinks to signal that they can report
+ * metrics.
+ *
+ * @since 3.3.0


It seems to be missed. Could you please update this?

jerrypeng · 2022-03-21T06:52:17Z

@HeartSaVioR thanks for the review again! PTAL!

HeartSaVioR · 2022-03-21T07:24:27Z

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

-                        mode: SaveMode,
-                        parameters: Map[String, String],
-                        data: DataFrame): BaseRelation = {
+    def createRelation(sqlContext: SQLContext,


probably the last nit: the indentation rule Spark uses is quite different with others, hence you'd like to refer to the Scala style guide.

https://github.com/databricks/scala-style-guide

https://github.com/databricks/scala-style-guide#spacing-and-indentation

For method declarations, use 4 space indentation for their parameters and put each in each line when the parameters don't fit in two lines. Return types can be either on the same line as the last parameter, or start a new line with 2 space indent.

Below is the correct indentation for this case.

def createRelation( sqlContext: SQLContext, mode: SaveMode, parameters: Map[String, String], data: DataFrame): BaseRelation = {

jerrypeng · 2022-03-21T21:41:41Z

@HeartSaVioR thanks for the review again! PTAL!

HeartSaVioR

+1

HeartSaVioR · 2022-03-22T06:35:32Z

Thanks! Merging to master!

HyukjinKwon · 2022-03-23T03:07:53Z

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

+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+
+class ReportSinkMetricsSuite extends StreamTest {


The tests added here seem flaky:

ReportSinkMetricsSuite: - test ReportSinkMetrics *** FAILED *** (244 milliseconds) Expected null, but got {"metrics-1"="value-1", "metrics-2"="value-2"} (ReportSinkMetricsSuite.scala:75) org.scalatest.exceptions.TestFailedException: at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472) at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471) at org.scalatest.funsuite.AnyFunSuite.newAssertionFailedException(AnyFunSuite.scala:1563) at org.scalatest.Assertions.assertResult(Assertions.scala:867) at org.scalatest.Assertions.assertResult$(Assertions.scala:863) at org.scalatest.funsuite.AnyFunSuite.assertResult(AnyFunSuite.scala:1563) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.$anonfun$new$2(ReportSinkMetricsSuite.scala:75) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.$anonfun$new$2$adapted(ReportSinkMetricsSuite.scala:60) at org.apache.spark.sql.test.SQLTestUtils.$anonfun$withTempDir$1(SQLTestUtils.scala:79) at org.apache.spark.sql.test.SQLTestUtils.$anonfun$withTempDir$1$adapted(SQLTestUtils.scala:78) at org.apache.spark.SparkFunSuite.withTempDir(SparkFunSuite.scala:221) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.org$apache$spark$sql$test$SQLTestUtils$$super$withTempDir(ReportSinkMetricsSuite.scala:35) at org.apache.spark.sql.test.SQLTestUtils.withTempDir(SQLTestUtils.scala:78) at org.apache.spark.sql.test.SQLTestUtils.withTempDir$(SQLTestUtils.scala:77) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.withTempDir(ReportSinkMetricsSuite.scala:35) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.$anonfun$new$1(ReportSinkMetricsSuite.scala:60) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)

https://github.com/apache/spark/runs/5646670314?check_suite_focus=true

Thanks for reporting. Seems odd. @jerrypeng Could you please check this?

Actually, the fix seems pretty simple .. I made a quick followup- #35945

I was a bit late lol https://github.com/apache/spark/pull/35872/files#r832810740

HeartSaVioR · 2022-03-23T03:18:03Z

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

+          query.processAllAvailable()
+        }
+
+        assertResult(metricsMap) {


OK I think we missed that listener callback happens in different thread than stream thread.

@jerrypeng
We may need to add sc.listenerBus.waitUntilEmpty(), or wrap this with eventually. Could you please create a follow-up PR? Thanks in advance!

…csSuite ### What changes were proposed in this pull request? The test is flaky: ``` ReportSinkMetricsSuite: - test ReportSinkMetrics *** FAILED *** (244 milliseconds) Expected null, but got {"metrics-1"="value-1", "metrics-2"="value-2"} (ReportSinkMetricsSuite.scala:75) org.scalatest.exceptions.TestFailedException: at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472) at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471) at org.scalatest.funsuite.AnyFunSuite.newAssertionFailedException(AnyFunSuite.scala:1563) at org.scalatest.Assertions.assertResult(Assertions.scala:867) at org.scalatest.Assertions.assertResult$(Assertions.scala:863) at org.scalatest.funsuite.AnyFunSuite.assertResult(AnyFunSuite.scala:1563) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.$anonfun$new$2(ReportSinkMetricsSuite.scala:75) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.$anonfun$new$2$adapted(ReportSinkMetricsSuite.scala:60) at org.apache.spark.sql.test.SQLTestUtils.$anonfun$withTempDir$1(SQLTestUtils.scala:79) at org.apache.spark.sql.test.SQLTestUtils.$anonfun$withTempDir$1$adapted(SQLTestUtils.scala:78) at org.apache.spark.SparkFunSuite.withTempDir(SparkFunSuite.scala:221) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.org$apache$spark$sql$test$SQLTestUtils$$super$withTempDir(ReportSinkMetricsSuite.scala:35) at org.apache.spark.sql.test.SQLTestUtils.withTempDir(SQLTestUtils.scala:78) at org.apache.spark.sql.test.SQLTestUtils.withTempDir$(SQLTestUtils.scala:77) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.withTempDir(ReportSinkMetricsSuite.scala:35) at org.apache.spark.sql.streaming.ReportSinkMetricsSuite.$anonfun$new$1(ReportSinkMetricsSuite.scala:60) at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) ``` We should wait all events to be processed. See #35872 (comment). ### Why are the changes needed? To make the test not flaky. ### Does this PR introduce _any_ user-facing change? No, test-only. ### How was this patch tested? Existing tests. CI in this PR should test it out. Closes #35945 from HyukjinKwon/SPARK-38564. Authored-by: Hyukjin Kwon <gurwls223@apache.org> Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

[SPARK-38564] Support collecting metrics from streaming sinks

9a9ee09

github-actions bot added SQL STRUCTURED STREAMING labels Mar 16, 2022

HyukjinKwon changed the title ~~[SPARK-38564] Support collecting metrics from streaming sinks~~ [SPARK-38564][ Support collecting metrics from streaming sinks Mar 17, 2022

HyukjinKwon changed the title ~~[SPARK-38564][ Support collecting metrics from streaming sinks~~ [SPARK-38564][SS] Support collecting metrics from streaming sinks Mar 17, 2022

HeartSaVioR reviewed Mar 18, 2022

View reviewed changes

add test

3d86963

jerrypeng requested a review from HeartSaVioR March 20, 2022 08:41

reformat

6aaf007

HeartSaVioR reviewed Mar 20, 2022

View reviewed changes

addressing comments

aa73fb0

jerrypeng requested a review from HeartSaVioR March 21, 2022 06:52

HeartSaVioR reviewed Mar 21, 2022

View reviewed changes

fix formatting

cbec88b

jerrypeng requested a review from HeartSaVioR March 21, 2022 21:41

HeartSaVioR approved these changes Mar 22, 2022

View reviewed changes

HeartSaVioR closed this in fc5e922 Mar 22, 2022

HyukjinKwon reviewed Mar 23, 2022

View reviewed changes

HyukjinKwon mentioned this pull request Mar 23, 2022

[SPARK-38564][SS][TESTS] Wait all events to arrive in ReportSinkMetricsSuite #35945

Closed

HeartSaVioR reviewed Mar 23, 2022

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-38564][SS] Support collecting metrics from streaming sinks #35872

[SPARK-38564][SS] Support collecting metrics from streaming sinks #35872

jerrypeng commented Mar 16, 2022 •

edited by HeartSaVioR

Loading

HyukjinKwon commented Mar 17, 2022

AmplabJenkins commented Mar 17, 2022

HeartSaVioR left a comment

HeartSaVioR Mar 18, 2022

HeartSaVioR Mar 20, 2022

jerrypeng Mar 21, 2022

jerrypeng commented Mar 20, 2022

HeartSaVioR Mar 20, 2022

jerrypeng Mar 21, 2022

HeartSaVioR Mar 20, 2022

jerrypeng Mar 21, 2022

HeartSaVioR Mar 20, 2022

jerrypeng Mar 21, 2022

HeartSaVioR Mar 20, 2022

jerrypeng commented Mar 21, 2022

HeartSaVioR Mar 21, 2022

jerrypeng Mar 21, 2022

jerrypeng commented Mar 21, 2022

HeartSaVioR left a comment

HeartSaVioR commented Mar 22, 2022

HyukjinKwon Mar 23, 2022

HeartSaVioR Mar 23, 2022

HyukjinKwon Mar 23, 2022

HeartSaVioR Mar 23, 2022

HeartSaVioR Mar 23, 2022


		inputData.addData(1, 2, 3)

		var metricsMap: java.util.Map[String, String] = null

[SPARK-38564][SS] Support collecting metrics from streaming sinks #35872

[SPARK-38564][SS] Support collecting metrics from streaming sinks #35872

Conversation

jerrypeng commented Mar 16, 2022 • edited by HeartSaVioR Loading

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?

HyukjinKwon commented Mar 17, 2022

AmplabJenkins commented Mar 17, 2022

HeartSaVioR left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerrypeng commented Mar 20, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerrypeng commented Mar 21, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerrypeng commented Mar 21, 2022

HeartSaVioR left a comment

Choose a reason for hiding this comment

HeartSaVioR commented Mar 22, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerrypeng commented Mar 16, 2022 •

edited by HeartSaVioR

Loading