[SPARK-29348][SQL][FOLLOWUP] Fix slight bug on streaming example for Dataset.observe

[HeartSaVioR]

What changes were proposed in this pull request?

This patch fixes a small bug in the example of streaming query, as the type of observable metrics is Java Map instead of Scala Map, so to use foreach it should be converted first.

Why are the changes needed?

Described above.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Ran below query via spark-shell:

Streaming

import scala.collection.JavaConverters._
import scala.util.Random
import org.apache.spark.sql.streaming.StreamingQueryListener
import org.apache.spark.sql.streaming.StreamingQueryListener._

spark.streams.addListener(new StreamingQueryListener() {
  override def onQueryProgress(event: QueryProgressEvent): Unit = {
    event.progress.observedMetrics.asScala.get("my_event").foreach { row =>
      // Trigger if the number of errors exceeds 5 percent
      val num_rows = row.getAs[Long]("rc")
      val num_error_rows = row.getAs[Long]("erc")
      val ratio = num_error_rows.toDouble / num_rows
      if (ratio > 0.05) {
        // Trigger alert
        println(s"alert! error ratio: $ratio")
      }
    }
  }

  def onQueryStarted(event: QueryStartedEvent): Unit = {}
  def onQueryTerminated(event: QueryTerminatedEvent): Unit = {}
})

val rates = spark
  .readStream
  .format("rate")
  .option("rowsPerSecond", 10)
  .load

val rand = new Random()
val df = rates.map { row => (row.getLong(1), if (row.getLong(1) % 2 == 0) "error" else null) }.toDF
val ds = df.selectExpr("_1 AS id", "_2 AS error")
// Observe row count (rc) and error row count (erc) in the batch Dataset
val observed_ds = ds.observe("my_event", count(lit(1)).as("rc"), count($"error").as("erc"))
observed_ds.writeStream.format("console").start()

[HeartSaVioR]

This was a minor bug on example: unlike observedMetrics available in QueryExecutionListener, event.progress.observedMetrics is a "Java" Map instead of "Scala" Map, hence to use foreach it needs asScala to convert to Scala Map.

[srowen]

Sounds fine. The alternative is to wrap it in Option(....get(...)).foreach but that's messier

[HeartSaVioR]

Thanks for reviewing, reflected review comments, as well as updated PR description to reflect the change.

[HyukjinKwon]

Looks good by me

[SparkQA]

Test build #115935 has finished for PR 27046 at commit 6dfa13f.

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

[SparkQA]

Test build #115931 has finished for PR 27046 at commit 8398cf4.

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

[SparkQA]

Test build #115936 has finished for PR 27046 at commit 4c16552.

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

[HyukjinKwon]

retest this please

[HyukjinKwon]

sorry last nit. this one too.

[HyukjinKwon]

FYI @hvanhovell as I believe you're already using this API; this changes the method signature.

[hvanhovell]

See my comments on this in #26127. I am -1 on this in its current form. I am fine with fixing the example.

[HeartSaVioR]

For the record, the link of key comment is here: #26127 (comment)

[HeartSaVioR]

Just updated. Please take a look again. Thanks in advance!

[SparkQA]

Test build #115940 has finished for PR 27046 at commit 4c16552.

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

[HyukjinKwon]

Discussed with @hvanhovell offline as well. Yup, I am good with it.

[SparkQA]

Test build #115954 has finished for PR 27046 at commit 2878cdb.

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

[HyukjinKwon]

Merged to master.

[HeartSaVioR]

Thanks for reviewing and merging!