Konfigurace úložiště stavů RocksDB v Azure Databricks

RocksDB je výchozí zprostředkovatel úložiště stavů v Databricks Runtime 17.3 a novější. Pro databricks Runtime verze nižší než 17.3 můžete povolit správu stavu založené na RocksDB nastavením následující konfigurace ve SparkSession před spuštěním dotazu streamování.

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

RocksDB můžete povolit v deklarativních kanálech Sparku pro Lakeflow. Vizte Optimalizujte konfiguraci pipeliny pro stavové zpracování.

Povolení vytváření kontrolních bodů protokolu změn

Ve službě Databricks Runtime 13.3 LTS a novějších můžete povolit kontrolu bodů protokolu změn, aby se snížila doba trvání kontrolního bodu a celková latence u úloh strukturovaného streamování. Databricks doporučuje povolit kontrolu verzí pro všechny stavové dotazy v rámci strukturovaného streamování. Vytváření kontrolních bodů v protokolu změn je ve výchozím nastavení povolené v Databricks Runtime 17.3 a novějších verzích.

Tradičně RocksDB State Store vytváří snímky a nahrává datové soubory během provádění kontrolních bodů. Aby se těmto nákladům předešlo, při vytváření kontrolních bodů protokolu změn se do trvalého úložiště zapisují pouze záznamy, které se od posledního kontrolního bodu změnily.

Vytváření kontrolních bodů protokolu změn je ve výchozím nastavení zakázané ve verzích Databricks Runtime nižších než 17.3. Kontrolní body protokolu změn můžete povolit na úrovni SparkSession pomocí následující syntaxe:

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

U existujícího datového proudu můžete povolit kontrolní body protokolu změn a udržovat informace o stavu uložené v kontrolním bodu.

Důležité

Dotazy, kde je zapnuto vytváření kontrolních bodů protokolu změn, lze spouštět pouze na Databricks Runtime 13.3 LTS a novějších verzích. Pokud zakážete vytváření kontrolních bodů changelogu, čímž se vrátíte ke staršímu stylu kontrolních bodů, musíte i nadále spouštět tyto dotazy v Databricks Runtime 13.3 LTS nebo vyšší. Aby se tyto změny projevily, musíte úlohu restartovat.

Metriky úložiště stavu RocksDB

Každý stavový operátor shromažďuje metriky spojené s operacemi správy stavu prováděnými na jeho instanci RocksDB, aby bylo možné sledovat úložiště stavu a potenciálně pomoci s laděním pomalosti úloh.

V prostředí Databricks Runtime 16.4 LTS a vyšším jsou metriky pro konkrétní stavová úložiště označeny jejich ID oddílu a názvem úložiště, aby zajistily jejich oddělení. Všechny ostatní metriky jsou hlášeny jako agregační součet pro všechny operátory stavu ve všech úkolech, ve kterých je operátor stavu spuštěný.

Tyto metriky jsou součástí customMetrics mapy uvnitř stateOperators polí v StreamingQueryProgress. Následuje příklad StreamingQueryProgress ve formátu JSON (získaném pomocí 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": {}
}

Podrobné popisy metrik jsou následující:

Název metriky	Popis
Latence potvrzení dávky zápisu RocksDB	Doba (v milisekundách) potřebná k použití fázovaných zápisů do struktury v paměti (WriteBatch) do nativní RocksDB.
rocksdbCommitFlushLatency	Čas (v milis) trvalo vyprázdnění změn v paměti RocksDB na místní disk.
rocksdbCommitCompactLatency	Čas (v milisekundách) potřebný pro kompakci (volitelné) během potvrzení kontrolního bodu.
rocksdbCommitPauseLatency	Čas (v milisekundách) potřebný k zastavení pracovních vláken na pozadí (například pro optimalizaci komprese) během potvrzení kontrolního bodu.
LatenceCommituKontrolníhoBoduRocksDB	Čas (v milisekundách) potřebný na pořízení snímku nativního RocksDB a jeho zápis do místního adresáře.
rocksdbCommitFileSyncLatenceMs	Čas (v milisekundách), který byl potřeba k synchronizaci nativních souborů spojených se snímky RocksDB na externí úložiště, místo kontrolního bodu.
rocksdbGetLatency	Průměrná doba (v nanosekundách) připadala na základní nativní RocksDB::Get volání.
rocksdbPutCount	Průměrná doba (v nanosekundách) připadala na základní nativní RocksDB::Put volání.
rocksdbGetCount	Počet nativních RocksDB::Get volání (nezahrnuje Gets volání z WriteBatch – dávky v paměti používané pro dočasné zápisy).
rocksdbPutCount	Počet nativních RocksDB::Put volání (nezahrnuje Puts WriteBatch – dávku uchovávanou v paměti pro přípravu zápisů).
rocksdbCelkovýPočetBajtůPřečtenýchGetem	Počet nekomprimovaných bajtů přečtených prostřednictvím nativních RocksDB::Get volání.
rocksdbCelkemBajtůZapsaných"Put"	Počet nekomprimovaných bajtů zapsaných prostřednictvím nativních RocksDB::Put volání
rocksdbReadBlockCacheHitCount	Počet použití nativní mezipaměti bloků RocksDB k tomu, aby se zabránilo čtení dat z místního disku.
rocksdbPočetChybČteníBlokuMezipaměti	Počet případů, kdy chyběla data v nativní blokové mezipaměti RocksDB a bylo nutné číst data z místního disku.
rocksdbCelkovýPočetBytůPřečtenýchKompresí	Počet bajtů načtených z místního disku nativním procesem komprimace RocksDB
rocksdbCelkemBajtůNapsanýchKompakcí	Počet bajtů zapsaných na místní disk nativním procesem komprimace RocksDB
rocksdbCelkováLatenceKompakceMs	Doba (v milisekundách), kterou trvala komprimace RocksDB (jak na pozadí, tak i volitelná komprimace, která byla zahájena během potvrzování).
rocksdbWriterStallLatencyMs	Čas (v milisekundách), kdy se zapisovač zastavil kvůli kompresi na pozadí nebo vyprázdnění memtables na disk.
rocksdbCelkovéBajtyPřečtenéPřesIterator	Některé stavové operace (například zpracování časového limitu v flatMapGroupsWithState nebo vodoznakování v agregacích s okny) vyžadují čtení veškerých dat v databázi prostřednictvím iterátoru. Celková velikost nekomprimovaných dat přečtených pomocí iterátoru.

Omezit využití paměti RocksDB

V Databricks Runtime 17.3 a vyšších Azure Databricks automaticky omezuje využití paměti RocksDB na uzel. Ve starších verzích databricks Runtime nakonfigurujte tuto možnost ručně, aby se zabránilo chybám nedostatku paměti.

RocksDB přiděluje paměť pro memtables, mezipaměť bloků a filtrovací a indexové bloky. Bez omezení může využití paměti napříč několika instancemi RocksDB na uzlu narůstat neomezeně dlouho a způsobit chyby kvůli nedostatku paměti. Správce vyrovnávací paměti zápisu RocksDB prosazuje limit paměti na uzel napříč všemi instancemi RocksDB běžícími na daném uzlu.

Pokud chcete využití paměti RocksDB limitovat, nastavte před spuštěním dotazu streamování následující konfigurace v relaci Sparku:

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

Následující konfigurace řídí využití paměti RocksDB:

Konfigurace	Popis
spark.sql.streaming.stateStore.rocksdb.boundedMemoryUsage	Povolí omezení sdílené paměti ve všech instancích RocksDB na uzlu. Nastavte na true, chcete-li povolit.
spark.sql.streaming.stateStore.rocksdb.maxMemoryUsageMB	Maximální paměť v MB sdílená napříč všemi instancemi RocksDB na uzlu Nastavte ji na statickou hodnotu nebo na hodnotu vypočítanou jako zlomek fyzické paměti uzlu.
spark.sql.streaming.stateStore.rocksdb.writeBufferSizeMB	Maximální velikost vyrovnávací paměti zápisu v MB pro jednotlivé instance RocksDB. Ve výchozím nastavení se používá interní hodnota RocksDB.
spark.sql.streaming.stateStore.rocksdb.maxWriteBufferNumber	Maximální počet vyrovnávacích pamětí zápisu pro jednotlivé instance RocksDB. Ve výchozím nastavení se použije interní hodnota databáze RocksDB.