Konfigurieren des RocksDB-Statusspeichers auf Azure Databricks

  • Artikel

Sie können die RocksDB-basierte Zustandsverwaltung aktivieren, indem Sie die folgende Konfiguration in der SparkSession-Instanz festlegen, bevor Sie die Streamingabfrage starten.

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

Sie können RocksDB für Delta Live Tables-Pipelines aktivieren. Weitere Informationen finden Sie unter Optimieren der Pipelinekonfiguration für die zustandsbehaftete Verarbeitung.

Aktivieren der Änderungsprotokollprüfpunkte

In Databricks Runtime 13.3 LTS und höher können Sie Prüfpunkte im Änderungsprotokoll aktivieren, um die Prüfpunktdauer und End-to-End-Latenz für strukturierte Streamingworkloads zu verringern. Databricks empfiehlt, Änderungsprotokollprüfpunkte für alle zustandsbehafteten Abfragen von strukturiertem Streaming zu aktivieren.

Normalerweise werden Momentaufnahmen des RocksDB-Zustandsspeichers während des Prüfpunkts erstellt und Datendateien hochgeladen. Um diese Kosten zu vermeiden, schreiben Änderungsprotokollprüfpunkte nur Datensätze, die sich seit dem letzten Prüfpunkt geändert haben, in dauerhaften Speicher."

Änderungsprotokollprüfpunkte sind standardmäßig deaktiviert. Sie können Änderungsprotokollprüfpunkte in der SparkSession-Ebene mit der folgenden Syntax aktivieren:

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

Sie können die Änderungsprotokollprüfpunkte für einen vorhandenen Stream aktivieren und die im Prüfpunkt gespeicherten Zustandsinformationen verwalten.

Wichtig

Abfragen mit aktivierten Prüfpunkten im Änderungsprotokoll, können nur unter Databricks Runtime 13.3 LTS und höher ausgeführt werden. Sie können Prüfpunkte im Änderungsprotokoll deaktivieren, um das Legacyverhalten von Prüfpunkten wiederherzustellen. Sie müssen diese Abfragen jedoch trotzdem in Databricks Runtime 13.3 LTS oder höher ausführen. Sie müssen den Auftrag neu starten, damit diese Änderungen vorgenommen werden.

Metriken des RocksDB-Zustandsspeichers

Jeder Zustandsoperator erfasst Metriken im Zusammenhang mit den Zustandsverwaltungsvorgängen, die auf seiner RocksDB-Instanz ausgeführt werden, um den Zustandsspeicher zu beobachten und möglicherweise bei der Behebung einer langsamen Auftragsausführung zu helfen. Diese Metriken werden pro Zustandsoperator im Auftrag für alle Aufgaben aggregiert (Summe), bei denen der Zustandsoperator ausgeführt wird. Diese Metriken sind Teil der customMetrics-Zuordnung in den stateOperators-Feldern in StreamingQueryProgress. Nachfolgend sehen Sie ein Beispiel für StreamingQueryProgress im JSON-Format (mit StreamingQueryProgress.json() abgerufen).

{
  "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" : {
  }
}

Ausführliche Beschreibungen der Metriken:

Metrikname BESCHREIBUNG
rocksdbCommitWriteBatchLatency Zeit (in Millisekunden) zum Anwenden der Stagingschreibvorgänge in der In-Memory-Struktur (WriteBatch) auf die native RocksDB-Instanz.
rocksdbCommitFlushLatency Zeit (in Millisekunden) für das Leeren der In-Memory-Änderungen in RocksDB auf den lokalen Datenträger.
rocksdbCommitCompactLatency Zeit (in Millisekunden) für die Komprimierung (optional) während des Prüfpunktcommits.
rocksdbCommitPauseLatency Zeit (in Millisekunden) zum Beenden der Arbeitsthreads im Hintergrund (für Komprimierung usw.) als Teil des Prüfpunkt-Commits.
rocksdbCommitCheckpointLatency Zeit (in Millisekunden) für das Erstellen einer Momentaufnahme der nativen RocksDB-Instanz und das Schreiben in ein lokales Verzeichnis.
rocksdbCommitFileSyncLatencyMs Zeit (in Millisekunden) für die Synchronisierung der Dateien für die Momentaufnahme der nativen RocksDB-Instanz mit einem externen Speicher (Prüfpunktspeicherort).
rocksdbGetLatency Durchschnittliche Zeit (in Nanosekunden) pro zugrunde liegendem nativen RocksDB::Get-Aufruf.
rocksdbPutCount Durchschnittliche Zeit (in Nanosekunden) pro zugrunde liegendem nativen RocksDB::Put-Aufruf.
rocksdbGetCount Anzahl nativer RocksDB::Get-Aufrufe (ohne Gets aus WriteBatch – In-Memory-Batch für Stagingschreibvorgänge verwendet).
rocksdbPutCount Anzahl nativer RocksDB::Put-Aufrufe (ohne Puts in WriteBatch – In-Memory-Batch für Stagingschreibvorgänge verwendet).
rocksdbTotalBytesReadByGet Anzahl unkomprimierter Bytes, die durch native RocksDB::Get-Aufrufe gelesen wurden.
rocksdbTotalBytesWrittenByPut Anzahl unkomprimierter Bytes, die durch native RocksDB::Put-Aufrufe geschrieben wurden.
rocksdbReadBlockCacheHitCount Anzahl, wie oft der native RocksDB-Blockcache verwendet wird, um das Lesen von Daten vom lokalen Datenträger zu vermeiden.
rocksdbReadBlockCacheMissCount Anzahl, wie oft der native RocksDB-Blockcache nicht funktioniert hat und Daten vom lokalen Datenträger gelesen werden mussten.
rocksdbTotalBytesReadByCompaction Anzahl der Bytes, die durch den nativen RocksDB-Komprimierungsprozess vom lokalen Datenträger gelesen wurden.
rocksdbTotalBytesWrittenByCompaction Anzahl der Bytes, die durch den nativen RocksDB-Komprimierungsprozess auf den lokalen Datenträger geschrieben wurden.
rocksdbTotalCompactionLatencyMs Zeit (in Millisekunden) für RocksDB-Komprimierungen (sowohl im Hintergrund als auch optionale Komprimierung, die während des Commits initiiert wurde).
rocksdbWriterStallLatencyMs Zeit (in Millisekunden), über die der Writer aufgrund einer Komprimierung im Hintergrund oder zum Leeren der Memtables auf den Datenträger angehalten hat.
rocksdbTotalBytesReadThroughIterator Einige zustandsbehaftete Vorgänge (z. B. Timeoutverarbeitung in flatMapGroupsWithState oder Wasserzeichen für Aggregationen im Fenstermodus) erfordern das Lesen der gesamten Daten in der Datenbank über Iterator. Die Gesamtgröße der mit dem Iterator gelesenen unkomprimierten Daten.