[GitHub] [spark] viirya commented on a change in pull request #32330: [SPARK-35215][SQL] Update custom metric per certain rows and at the end of the task

[GitBox <gi...@apache.org>]

viirya commented on a change in pull request #32330:
URL: https://github.com/apache/spark/pull/32330#discussion_r620012160



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousDataSourceRDD.scala
##########
@@ -92,10 +92,18 @@ class ContinuousDataSourceRDD(
 
     val partitionReader = readerForPartition.getPartitionReader()
     new NextIterator[InternalRow] {
+      private var numRow = 0L
+
       override def getNext(): InternalRow = {
-        partitionReader.currentMetricsValues.foreach { metric =>
-          customMetrics(metric.name()).set(metric.value())
+        if (numRow % CustomMetrics.numRowsPerUpdate == 0) {
+          partitionReader.currentMetricsValues.foreach { metric =>
+            assert(customMetrics.contains(metric.name()),

Review comment:
       I can remove it. I also thought it is not necessary but just added for a comment before.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/metric/CustomMetrics.scala
##########
@@ -25,6 +25,8 @@ import org.apache.spark.sql.connector.CustomMetric
 object CustomMetrics {
   private[spark] val V2_CUSTOM = "v2Custom"
 
+  private[spark] val numRowsPerUpdate = 100L

Review comment:
       `numRow` is a long, I guess this can be just int.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala
##########
@@ -92,12 +104,15 @@ private class PartitionIterator[T](
     if (!hasNext) {
       throw QueryExecutionErrors.endOfStreamError()
     }
-    reader.currentMetricsValues.foreach { metric =>
-      assert(customMetrics.contains(metric.name()),
-        s"Custom metrics ${customMetrics.keys.mkString(", ")} do not contain the metric " +
-          s"${metric.name()}")
-      customMetrics(metric.name()).set(metric.value())
+    if (numRow % CustomMetrics.numRowsPerUpdate == 0) {
+      reader.currentMetricsValues.foreach { metric =>

Review comment:
       sure.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org