viirya commented on a change in pull request #33455:
URL: https://github.com/apache/spark/pull/33455#discussion_r673774442
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBStateStoreProvider.scala
##########
@@ -215,40 +239,79 @@ object RocksDBStateStoreProvider {
// Native operation latencies report as latency per 1000 calls
// as SQLMetrics support ms latency whereas RocksDB reports it in
microseconds.
- val CUSTOM_METRIC_GET_TIME = StateStoreCustomTimingMetric(
- "rocksdbGetLatency", "RocksDB: avg get latency (per 1000 calls)")
- val CUSTOM_METRIC_PUT_TIME = StateStoreCustomTimingMetric(
- "rocksdbPutLatency", "RocksDB: avg put latency (per 1000 calls)")
+ val CUSTOM_METRIC_GET_TIME = StateStoreCustomTimingNsMetric(
+ "rocksdbGetLatency", "RocksDB: avg get latency")
+ val CUSTOM_METRIC_PUT_TIME = StateStoreCustomTimingNsMetric(
+ "rocksdbPutLatency", "RocksDB: avg put latency")
+
+ val CUSTOM_METRIC_GET_COUNT = StateStoreCustomSumMetric(
+ "rocksdbGetCount", "RocksDB: number of get calls")
+ val CUSTOM_METRIC_PUT_COUNT = StateStoreCustomSumMetric(
+ "rocksdbPutCount", "RocksDB: number of put calls")
// Commit latency detailed breakdown
val CUSTOM_METRIC_WRITEBATCH_TIME = StateStoreCustomTimingMetric(
"rocksdbCommitWriteBatchLatency", "RocksDB: commit - write batch time")
val CUSTOM_METRIC_FLUSH_TIME = StateStoreCustomTimingMetric(
"rocksdbCommitFlushLatency", "RocksDB: commit - flush time")
+ val CUSTOM_METRIC_COMMIT_COMPACT_TIME = StateStoreCustomTimingMetric(
+ "rocksdbCommitCompactLatency", "RocksDB: commit - compact time")
val CUSTOM_METRIC_PAUSE_TIME = StateStoreCustomTimingMetric(
"rocksdbCommitPauseLatency", "RocksDB: commit - pause bg time")
val CUSTOM_METRIC_CHECKPOINT_TIME = StateStoreCustomTimingMetric(
"rocksdbCommitCheckpointLatency", "RocksDB: commit - checkpoint time")
val CUSTOM_METRIC_FILESYNC_TIME = StateStoreCustomTimingMetric(
- "rocksdbFileSyncTime", "RocksDB: commit - file sync time")
- val CUSTOM_METRIC_FILES_COPIED = StateStoreCustomSizeMetric(
+ "rocksdbCommitFileSyncLatencyMs", "RocksDB: commit - file sync to external
storage time")
+ val CUSTOM_METRIC_FILES_COPIED = StateStoreCustomSumMetric(
"rocksdbFilesCopied", "RocksDB: file manager - files copied")
val CUSTOM_METRIC_BYTES_COPIED = StateStoreCustomSizeMetric(
"rocksdbBytesCopied", "RocksDB: file manager - bytes copied")
- val CUSTOM_METRIC_FILES_REUSED = StateStoreCustomSizeMetric(
+ val CUSTOM_METRIC_FILES_REUSED = StateStoreCustomSumMetric(
"rocksdbFilesReused", "RocksDB: file manager - files reused")
val CUSTOM_METRIC_ZIP_FILE_BYTES_UNCOMPRESSED = StateStoreCustomSizeMetric(
"rocksdbZipFileBytesUncompressed", "RocksDB: file manager - uncompressed
zip file bytes")
+ val CUSTOM_METRIC_BLOCK_CACHE_MISS = StateStoreCustomSumMetric(
+ "rocksdbReadBlockCacheMissCount",
+ "RocksDB: read - count of cache misses that required reading from local
disk")
+ val CUSTOM_METRIC_BLOCK_CACHE_HITS = StateStoreCustomSumMetric(
+ "rocksdbReadBlockCacheHitCount",
+ "RocksDB: read - count of cache hits in RocksDB block cache avoiding disk
read")
+ val CUSTOM_METRIC_BYTES_READ = StateStoreCustomSizeMetric(
+ "rocksdbTotalBytesReadByGet",
+ "RocksDB: read - total of uncompressed bytes read (from
memtables/cache/sst) from DB::Get()")
+ val CUSTOM_METRIC_BYTES_WRITTEN = StateStoreCustomSizeMetric(
+ "rocksdbTotalBytesWrittenByPut",
+ "RocksDB: write - total of uncompressed bytes written by DB::Put()")
Review comment:
DB::{Put(), Delete(), Merge(), Write()}?
--
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]
