在 Azure Databricks 上配置 RocksDB 状态存储

可以在启动流式处理查询之前，通过在 SparkSession 中设置以下配置来启用基于 RocksDB 的状态管理。

spark.conf.set(
  "spark.sql.streaming.stateStore.providerClass",
  "com.databricks.sql.streaming.state.RocksDBStateStoreProvider")

可以在增量实时表管道上启用 RocksDB。 请参阅针对有状态处理优化管道配置。

启用更改日志检查点

在 Databricks Runtime 13.3 LTS 及更高版本中，可以启用更改日志检查点功能，以降低结构化流工作负载的检查点持续时间和端到端延迟。 Databricks 建议为所有“结构化流式处理”有状态查询启用更改日志检查点。

传统来说，RocksDB 状态存储在检查点期间快照和上传数据文件。 为了避免这种成本，changelog 检查点仅将自上一个检查点以来更改的记录写入持久存储。”

默认情况下，更改日志检查点处于禁用状态。 可以使用以下语法在 SparkSession 级别启用更改日志检查点：

spark.conf.set(
  "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled", "true")

可以在现有流上启用更改日志检查点，并维护存储在检查点中的状态信息。

RocksDB 状态存储指标

每个状态运算符收集有关在其 RocksDB 实例上执行的状态管理操作的指标，这些指标可用于观测状态存储，并可能有助于调试作业速度缓慢问题。 这些指标将会根据运行状态运算符的所有任务中的每个作业中状态运算符进行聚合（加总）。 这些指标是 StreamingQueryProgress 中 stateOperators 字段内的 customMetrics 映射的一部分。 下面是采用 JSON 格式的 StreamingQueryProgress 的示例（使用 StreamingQueryProgress.json() 获取）。

{
  "id" : "6774075e-8869-454b-ad51-513be86cfd43",
  "runId" : "3d08104d-d1d4-4d1a-b21e-0b2e1fb871c5",
  "batchId" : 7,
  "stateOperators" : [ {
    "numRowsTotal" : 20000000,
    "numRowsUpdated" : 20000000,
    "memoryUsedBytes" : 31005397,
    "numRowsDroppedByWatermark" : 0,
    "customMetrics" : {
      "rocksdbBytesCopied" : 141037747,
      "rocksdbCommitCheckpointLatency" : 2,
      "rocksdbCommitCompactLatency" : 22061,
      "rocksdbCommitFileSyncLatencyMs" : 1710,
      "rocksdbCommitFlushLatency" : 19032,
      "rocksdbCommitPauseLatency" : 0,
      "rocksdbCommitWriteBatchLatency" : 56155,
      "rocksdbFilesCopied" : 2,
      "rocksdbFilesReused" : 0,
      "rocksdbGetCount" : 40000000,
      "rocksdbGetLatency" : 21834,
      "rocksdbPutCount" : 1,
      "rocksdbPutLatency" : 56155599000,
      "rocksdbReadBlockCacheHitCount" : 1988,
      "rocksdbReadBlockCacheMissCount" : 40341617,
      "rocksdbSstFileSize" : 141037747,
      "rocksdbTotalBytesReadByCompaction" : 336853375,
      "rocksdbTotalBytesReadByGet" : 680000000,
      "rocksdbTotalBytesReadThroughIterator" : 0,
      "rocksdbTotalBytesWrittenByCompaction" : 141037747,
      "rocksdbTotalBytesWrittenByPut" : 740000012,
      "rocksdbTotalCompactionLatencyMs" : 21949695000,
      "rocksdbWriterStallLatencyMs" : 0,
      "rocksdbZipFileBytesUncompressed" : 7038
    }
  } ],
  "sources" : [ {
  } ],
  "sink" : {
  }
}

指标的详细说明如下：