Skip to content
This repository was archived by the owner on Jan 9, 2020. It is now read-only.

Commit 1bc17a6

Browse files
Taaffysrowen
authored andcommitted
[SPARK-22052] Incorrect Metric assigned in MetricsReporter.scala
Current implementation for processingRate-total uses wrong metric: mistakenly uses inputRowsPerSecond instead of processedRowsPerSecond ## What changes were proposed in this pull request? Adjust processingRate-total from using inputRowsPerSecond to processedRowsPerSecond ## How was this patch tested? Built spark from source with proposed change and tested output with correct parameter. Before change the csv metrics file for inputRate-total and processingRate-total displayed the same values due to the error. After changing MetricsReporter.scala the processingRate-total csv file displayed the correct metric. <img width="963" alt="processed rows per second" src="https://user-images.githubusercontent.com/32072374/30554340-82eea12c-9ca4-11e7-8370-8168526ff9a2.png"> Please review http://spark.apache.org/contributing.html before opening a pull request. Author: Taaffy <[email protected]> Closes apache#19268 from Taaffy/patch-1.
1 parent 7c92351 commit 1bc17a6

File tree

1 file changed

+1
-1
lines changed

1 file changed

+1
-1
lines changed

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ class MetricsReporter(
4040
// Metric names should not have . in them, so that all the metrics of a query are identified
4141
// together in Ganglia as a single metric group
4242
registerGauge("inputRate-total", () => stream.lastProgress.inputRowsPerSecond)
43-
registerGauge("processingRate-total", () => stream.lastProgress.inputRowsPerSecond)
43+
registerGauge("processingRate-total", () => stream.lastProgress.processedRowsPerSecond)
4444
registerGauge("latency", () => stream.lastProgress.durationMs.get("triggerExecution").longValue())
4545

4646
private def registerGauge[T](name: String, f: () => T)(implicit num: Numeric[T]): Unit = {

0 commit comments

Comments
 (0)