Configurar o repositório de estado do RocksDB no Azure Databricks

O RocksDB é o provedor de armazenamento de estado padrão no Databricks Runtime 17.3 e superior. Para versões do Databricks Runtime abaixo da 17.3, você pode habilitar o gerenciamento de estado baseado no RocksDB definindo a seguinte configuração no SparkSession antes de iniciar a consulta de streaming.

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

Você pode habilitar o RocksDB no Lakeflow Spark Declarative Pipelines. Consulte Otimizar a configuração do pipeline para processamento com estado.

Habilitar o ponto de verificação do log de alterações

No Databricks Runtime 13.3 LTS e superior, você pode habilitar o ponto de verificação do log de alterações para reduzir a duração do ponto de verificação e a latência de ponta a ponta para cargas de trabalho de Streaming Estruturado. A Databricks recomenda habilitar o ponto de verificação do log de alterações para todas as consultas com estado de Streaming Estruturado. Ponto de verificação do log de alterações é habilitado por padrão no Databricks Runtime 17.3 e superior.

Tradicionalmente, o RocksDB State Store tira instantâneos e faz upload de arquivos de dados durante o ponto de verificação. Para evitar esse custo, o ponto de verificação do log de alterações grava apenas os registros que foram alterados desde o último ponto de verificação em um armazenamento durável.

O ponto de verificação do log de alterações é desabilitado por padrão no Databricks Runtime abaixo da versão 17.3. É possível utilizar a seguinte sintaxe para habilitar o ponto de verificação do log de alterações no nível do SparkSession:

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

Você pode habilitar o ponto de verificação do log de alterações em um stream existente e manter as informações de estado armazenadas no ponto de verificação.

Métricas de armazenamento de estado RocksDB

Cada operador de estado coleta métricas relacionadas às operações de gerenciamento de estado realizadas na sua instância RocksDB para observar o repositório de estado e, potencialmente, ajudar na depuração de lentidão do trabalho.

No Databricks Runtime 16.4 LTS e posteriores, as métricas de uma instância específica do repositório de estado são rotuladas com a ID de partição e o nome do repositório, garantindo que permaneçam separadas. Todas as outras métricas são relatadas como a soma agregada para cada operador de estado em todas as tarefas em que o operador de estado está em execução.

Essas métricas fazem parte do mapa customMetrics dentro dos campos stateOperators em StreamingQueryProgress. Veja a seguir um exemplo de StreamingQueryProgress no formato JSON (obtido 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": {
        "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": {}
}

Descrições detalhadas das métricas são as seguinte:

Uso de memória do Cap RocksDB

No Databricks Runtime 17.3 e superior, o Azure Databricks limita automaticamente o uso de memória do RocksDB por nó. Em versões anteriores do Databricks Runtime, configure isso manualmente para evitar erros fora de memória.

O RocksDB aloca memória para memtables, o cache de blocos e os blocos de filtro e de índice. Sem um limite, o uso de memória entre várias instâncias do RocksDB em um nó pode aumentar indefinidamente e causar erros de falta de memória. O gerenciador do buffer de gravação do RocksDB aplica um limite de memória por nó para todas as instâncias do RocksDB em execução nesse nó.

Para limitar o uso de memória do RocksDB, defina as seguintes configurações na sessão do Spark antes de iniciar a consulta de streaming:

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

As configurações a seguir controlam o uso de memória do RocksDB: