在 Azure Databricks 設定 RocksDB 狀態存放區

您可以在啟動串流查詢之前，先在SparkSession 中設定下列設定，以啟用RocksDB型狀態管理。

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

您可以在 Delta Live Tables 管線上啟用 RocksDB。 請參閱 優化管線組態以進行具狀態處理。

啟用變更記錄檢查點

在 Databricks Runtime 13.3 LTS 和更新版本中，您可以啟用變更記錄檢查點，以降低結構化串流工作負載的檢查點持續時間和端對端延遲。 Databricks 建議為所有結構化串流具狀態的查詢，啟用變更記錄檢查點。

傳統上，RocksDB 狀態存放區快照集，並在檢查點期間上傳數據檔。 為了避免這項成本，changelog 檢查點只會寫入自上次檢查點之後變更為永久性記憶體的記錄。

預設會停用 Changelog 檢查點。 您可以使用下列語法，在 SparkSession 層級中啟用變更記錄檢查點：

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

您可以在現有的數據流上啟用變更記錄檢查點，並維護儲存在檢查點中的狀態資訊。

RocksDB 狀態存放區計量

每個狀態運算子都會收集與其 RocksDB 實例上執行之狀態管理作業相關的計量，以觀察狀態存放區，並可能協助偵錯作業速度緩慢。 這些計量會匯總每個狀態運算符在作業中執行狀態運算子的所有工作。 這些計量是 中欄位內地圖的stateOperatorsStreamingQueryProgress一部分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" : {
  }
}

計量的詳細描述如下：