在 Azure Databricks 上配置 RocksDB 状态存储

RocksDB 是 Databricks Runtime 17.3 及更高版本中的默认状态存储提供程序。 对于低于 17.3 的 Databricks Runtime 版本，可以在启动流式处理查询之前在 SparkSession 中设置以下配置来启用基于 RocksDB 的状态管理。

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

可以在 Lakeflow Spark 声明性管道上启用 RocksDB。 请参阅针对有状态处理优化管道配置。

启用更改日志检查点

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

传统来说，RocksDB 状态存储在检查点期间快照和上传数据文件。 为避免这种开销，变更日志检查点仅将自上次检查点以来发生变更的记录写入持久存储。

默认情况下，Databricks Runtime 版本低于 17.3 时禁用更改日志检查点。 您可以在 SparkSession 级别使用以下语法启用变更日志检查点功能：

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

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

重要

已启用更改日志检查点功能的查询只能在 Databricks Runtime 13.3 LTS 及更高版本上运行。 你可以禁用更改日志检查点功能以还原旧有的检查点行为，但必须继续在 Databricks Runtime 13.3 LTS 或更高版本上运行这些查询。 必须重启作业才能完成这些更改。

RocksDB 状态存储指标

每个状态运算符都会收集关于其在 RocksDB 实例上执行的状态管理操作的指标，这些指标可以用于监控状态存储，并可能有助于解决作业执行速度缓慢的调试问题。

在 Databricks Runtime 16.4 LTS 及更高版本中，特定状态存储实例的指标标有其分区 ID 和存储名称，确保它们保持独立。 所有其他指标均报告为每个状态运算符在所有运行该状态运算符的任务中的聚合总和。

这些指标是 customMetrics 中 stateOperators 字段内的 StreamingQueryProgress 映射的一部分。 下面是采用 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": {
        "SnapshotLastUploaded.partition_0_default": 7,
        "SnapshotLastUploaded.partition_1_default": 7,
        "SnapshotLastUploaded.partition_2_default": 6,
        "SnapshotLastUploaded.partition_3_default": 6,
        "SnapshotLastUploaded.partition_4_default": -1,
        "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": {}
}

指标的详细说明如下：

指标名称 说明
rocksdbCommitWriteBatchLatency 将内存中结构中的分阶段写入 (WriteBatch) 应用于本机 RocksDB 所花费的时间（以毫秒为单位）。
rocksdbCommitFlushLatency 将 RocksDB 内存中更改刷新到本地磁盘所花费的时间（以毫秒为单位）。
rocksdbCommitCompactLatency 在检查点提交期间进行压缩（可选）所花费的时间（以毫秒为单位）。
rocksdbCommitPauseLatency 在检查点提交过程中停止后台工作线程（以实现压缩等目的）所花费的时间（以毫秒为单位）。
rocksdbCommitCheckpointLatency 创建本机 RocksDB 快照并将其写入本地目录所花费的时间（以毫秒为单位）。
rocksdbCommitFileSyncLatencyMs 将本机 RocksDB 快照相关的文件同步到外部存储（检查点位置）所花费的时间（以毫秒为单位）。
rocksdbGetLatency 每个基础本机 RocksDB::Get 调用平均花费的时间（以纳秒为单位）。
rocksdbPutCount 每个基础本机 RocksDB::Put 调用平均花费的时间（以纳秒为单位）。
rocksdbGetCount 本机 RocksDB::Get 调用的数量（不包括从 WriteBatch 执行的 Gets - 用于暂存写入的内存中批处理）。
rocksdbPutCount 本机 RocksDB::Put 调用数量（不包括对 WriteBatch 执行的 Puts - 用于暂存写入的内存中批处理）。
rocksdbTotalBytesReadByGet 本机函数 RocksDB::Get 调用读取的未压缩字节数。
rocksdbTotalBytesWrittenByPut 通过本机 RocksDB::Put 调用写入的未压缩字节数。
rocksdbReadBlockCacheHitCount 使用本机 RocksDB 块缓存避免从本地磁盘读取数据的次数。
rocksdbReadBlockCacheMissCount 从本地磁盘读取数据所需的未命中本机 RocksDB 块缓存次数。
rocksdbTotalBytesReadByCompaction 本机 RocksDB 压缩进程从本地磁盘读取的字节数。
rocksdbTotalBytesWrittenByCompaction RocksDB 本地重组进程写入本地磁盘的字节数。
rocksdbTotalCompactionLatencyMs RocksDB 压缩（后台压缩以及在提交期间启动的可选压缩）所花费的时间（以毫秒为单位）。
rocksdbWriterStallLatencyMs 由于后台压缩或将内存表刷新到磁盘而导致写入器停滞的时间（以毫秒为单位）。
rocksdbTotalBytesReadThroughIterator 某些有状态操作（例如 flatMapGroupsWithState 中的超时处理或窗口聚合中的水印处理）需要通过迭代器读取 DB 中的整个数据。 使用迭代器读取的未压缩数据的总大小。

Cap RocksDB 内存使用情况

在 Databricks Runtime 17.3 及更高版本中，Azure Databricks会自动限制每个节点的 RocksDB 内存使用量。 在较早版本的 Databricks Runtime 中，需要手动配置此项以避免内存溢出错误。

RocksDB 为内存表、块缓存以及过滤器块和索引块分配内存。 没有限制，节点上多个 RocksDB 实例上的内存使用量可能会无限期增长，并导致内存不足错误。 RocksDB 写入缓冲区管理器对节点上运行的所有 RocksDB 实例强制实施每节点内存限制。

若要限制 RocksDB 内存使用量，在启动流式处理查询之前，在 Spark 会话中设置以下配置：

spark.conf.set(
  "spark.sql.streaming.stateStore.rocksdb.boundedMemoryUsage", "true")
spark.conf.set(
  "spark.sql.streaming.stateStore.rocksdb.maxMemoryUsageMB", "<value>")

以下配置控制 RocksDB 内存使用情况：

配置 说明
spark.sql.streaming.stateStore.rocksdb.boundedMemoryUsage 在节点上的所有 RocksDB 实例上启用共享内存限制。 设置为 true 以启用。
spark.sql.streaming.stateStore.rocksdb.maxMemoryUsageMB 节点上所有 RocksDB 实例之间共享的最大内存（以 MB 为单位）。 将其设置为静态值，或者设置为按节点物理内存的一定比例计算得出的值。
spark.sql.streaming.stateStore.rocksdb.writeBufferSizeMB 单个 RocksDB 实例的最大写入缓冲区大小（以 MB 为单位）。 默认值为 RocksDB 内部值。
spark.sql.streaming.stateStore.rocksdb.maxWriteBufferNumber 单个 RocksDB 实例的最大写入缓冲区数。 默认值为 RocksDB 内部值。
  • Last updated on