Configurare l'archivio di stato RocksDB su Azure Databricks

È possibile abilitare la gestione dello stato basata su RocksDB impostando la configurazione seguente in SparkSession prima di avviare la query di streaming.

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

È possibile abilitare RocksDB nelle pipeline di tabelle live Delta. Vedere Ottimizzare la configurazione della pipeline per l'elaborazione con stato.

Abilitare il checkpoint del log delle modifiche

In Databricks Runtime 13.3 LTS e versioni successive è possibile abilitare il checkpoint del log delle modifiche per ridurre la durata del checkpoint e la latenza end-to-end per i carichi di lavoro Structured Streaming. Databricks raccomanda di abilitare il checkpoint del log delle modifiche per tutte le query con stato di Structured Streaming.

In genere, gli snapshot dell'archivio stati di RocksDB e caricano i file di dati durante il checkpoint. Per evitare questo costo, il checkpoint del log delle modifiche scrive solo i record che sono stati modificati dall'ultimo checkpoint nell'archiviazione durevole."

Il checkpoint del log delle modifiche è disabilitato per impostazione predefinita. È possibile abilitare il checkpoint del log delle modifiche nel livello SparkSession usando la sintassi seguente:

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

È possibile abilitare il checkpoint del log delle modifiche in un flusso esistente e mantenere le informazioni sullo stato archiviate nel checkpoint.

Metriche dell'archivio stati di RocksDB

Ogni operatore di stato raccoglie le metriche correlate alle operazioni di gestione dello stato eseguite nell'istanza di RocksDB per osservare l'archivio stati e potenzialmente contribuire al debug della lentezza del processo. Queste metriche vengono aggregate (somma) per operatore di stato nel processo in tutte le attività in cui è in esecuzione l'operatore di stato. Queste metriche fanno parte della customMetrics mappa all'interno dei stateOperators campi in StreamingQueryProgress. Di seguito è riportato un esempio di StreamingQueryProgress in formato JSON (ottenuto usando 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" : {
  }
}

Le descrizioni dettagliate delle metriche sono le seguenti: