vkorukanti commented on a change in pull request #33455:
URL: https://github.com/apache/spark/pull/33455#discussion_r674187214
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
##########
@@ -106,22 +106,46 @@ private[sql] class RocksDBStateStoreProvider
override def metrics: StateStoreMetrics = {
val rocksDBMetrics = rocksDB.metrics
def commitLatencyMs(typ: String): Long =
rocksDBMetrics.lastCommitLatencyMs.getOrElse(typ, 0L)
- def avgNativeOpsLatencyMs(typ: String): Long = {
-
rocksDBMetrics.nativeOpsLatencyMicros.get(typ).map(_.avg).getOrElse(0.0).toLong
+ def nativeOpsLatencyMillis(typ: String): Long = {
+ rocksDBMetrics.nativeOpsMetrics.get(typ).map(_ * 1000).getOrElse(0)
+ }
+ def avgNativeOpsLatencyNanos(typ: String): Long = {
+ (rocksDBMetrics.nativeOpsHistograms.get(typ).map(_.avg *
1000).getOrElse(0.0)).toLong
Review comment:
Thinking more about it, not sure if the avg makes sense anymore as we
aggregate (sum) across the tasks and aggregaing (sum) of avg may be misleading.
I think it should be the sum of the latency and couple this with the `get`
count, we should be able to get the average across all tasks. Let me know what
you think.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
