vkorukanti commented on a change in pull request #33455:
URL: https://github.com/apache/spark/pull/33455#discussion_r678371060
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
##########
@@ -106,22 +106,43 @@ 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 sumNativeOpsLatencyMillis(typ: String): Long = {
+ rocksDBMetrics.nativeOpsHistograms.get(typ).map(_.sum /
1000).getOrElse(0)
+ }
+ def nativeOpsCount(typ: String): Long = {
+ rocksDBMetrics.nativeOpsHistograms.get(typ).map(_.count).getOrElse(0)
+ }
Review comment:
`nativeOpsCount` is extracting the `count` from histogram based metrics.
`nativeOpsMetrics` is extracting the value from ticker based metrics.
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
##########
@@ -106,22 +106,43 @@ 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 sumNativeOpsLatencyMillis(typ: String): Long = {
+ rocksDBMetrics.nativeOpsHistograms.get(typ).map(_.sum /
1000).getOrElse(0)
+ }
+ def nativeOpsCount(typ: String): Long = {
+ rocksDBMetrics.nativeOpsHistograms.get(typ).map(_.count).getOrElse(0)
+ }
Review comment:
`nativeOpsCount` is extracting the `count` from histogram based metrics.
`nativeOpsMetrics` is extracting the value from ticker based metrics.
##########
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)
Review comment:
The ticker based metrics are not just `counts` of some operations, they
also capture the time. Example one is
[here](https://javadoc.io/static/org.rocksdb/rocksdbjni/6.4.6/org/rocksdb/TickerType.html#STALL_MICROS).
This is just a utility method to extract the time out of these ticker based
metrics.
--
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]
