Partilhar via

Configurar o armazenamento de estado do RocksDB no Azure Databricks

  • Artigo

Você pode habilitar o gerenciamento de estado baseado em 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 ativar RocksDB nos pipelines do Delta Live Tables. Consulte Otimizar a configuração do pipeline para processamento com monitoração de estado.

Ativar ponto de verificação do changelog

No Databricks Runtime 13.3 LTS e superior, você pode habilitar o ponto de verificação do changelog para reduzir a duração do ponto de verificação e a latência de ponta a ponta para cargas de trabalho do Structured Streaming. O Databricks recomenda ativar o ponto de verificação do registo de alterações para todas as consultas com monitorização de estado de Transmissão em Fluxo Estruturada.

Tradicionalmente, o RocksDB State Store faz snapshots e carrega arquivos de dados durante o checkpoint. Para evitar esse custo, o ponto de verificação do changelog grava apenas os registros que foram alterados desde o último ponto de verificação para armazenamento durável."

O ponto de verificação do changelog está desativado por padrão. Você pode habilitar o ponto de verificação do changelog no nível SparkSession usando a seguinte sintaxe:

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

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

Importante

As consultas que ativaram o ponto de verificação do changelog só podem ser executadas no Databricks Runtime 13.3 LTS e em versões posteriores. Você pode desabilitar o ponto de verificação do changelog para reverter para o comportamento de ponto de verificação herdado, mas deve continuar a executar essas consultas no Databricks Runtime 13.3 LTS ou superior. Você deve reiniciar o trabalho para que essas alterações ocorram.

Métricas do armazenamento de estado do RocksDB

Cada operador de estado recolhe métricas relacionadas com as operações de gestão de estado realizadas na sua instância de RocksDB para observar a memória de estado e, eventualmente, ajudar a diagnosticar a lentidão do trabalho.

As métricas para uma instância de armazenamento de estado específica são rotuladas com seu ID de partição e nome de armazenamento, garantindo que permaneçam separadas. Todas as outras métricas são agregadas (soma) por operador de estado em cada tarefa e processo onde o operador de estado está a ser executado.

Essas métricas são parte do mapa customMetrics dentro dos campos stateOperators em StreamingQueryProgress. A seguir está um exemplo de StreamingQueryProgress no formulário 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": {}
}

As descrições detalhadas das métricas são as seguintes:

Nome da métrica Descrição
rocksdbCommitWriteBatchLatency Tempo (em milissegundos) necessário para aplicar as gravações em etapas na estrutura em memória (WriteBatch) ao RocksDB nativo.
rocksdbCommitFlushLatency Tempo (em milis) necessário para liberar as alterações na memória do RocksDB para o disco local.
rocksdbCommitCompactLatency Tempo (em milissegundos) necessário para compactação (opcional) durante o registo do checkpoint.
rocksdbCommitPauseLatency Tempo (em milis) necessário para parar os threads de trabalho em segundo plano (para compactação, etc.) como parte da confirmação do ponto de verificação.
rocksdbCommitCheckpointLatência Tempo (em milissegundos) levado para capturar um instantâneo do RocksDB nativo e gravá-lo no diretório local.
rocksdbCommitFileSyncLatencyMs Tempo (em milis) necessário para sincronizar os arquivos nativos relacionados ao snapshot do RocksDB com um armazenamento externo (local do ponto de verificação).
rocksdbGetLatency Tempo médio (em nanos) levado por chamada nativa RocksDB::Get subjacente.
rocksdbPutCount Tempo médio (em nanos) levado por chamada nativa RocksDB::Put subjacente.
rocksdbGetCount Número de chamadas nativas RocksDB::Get (não inclui Gets de WriteBatch - lote na memória usado para pré-processamento de gravações).
rocksdbPutCount Número de chamadas nativas RocksDB::Put (não inclui Puts para WriteBatch - um lote em memória usado para organizar gravações em preparação).
rocksdbTotalBytesReadByGet Número de bytes não compactados lidos através de chamadas nativas RocksDB::Get.
BytesTotaisEscritosPorPutNoRocksDB Número de bytes não compactados gravados por meio de chamadas nativas RocksDB::Put .
rocksdbReadBlockCacheHitCount Número de vezes que o cache de bloco nativo do RocksDB é usado para evitar a leitura de dados do disco local.
rocksdbReadBlockCacheMissCount Número de vezes que o cache de bloco nativo do RocksDB falhou e requereu a leitura de dados do disco local.
rocksdbTotalBytesReadByCompaction Número de bytes lidos do disco local pelo processo de compactação nativo do RocksDB.
rocksdbTotalBytesWrittenByCompaction Número de bytes gravados no disco local pelo processo de compactação nativo do RocksDB.
rocksdbTotalCompactionLatencyMs Tempo (em milissegundos) necessário para as compactações do RocksDB (tanto em segundo plano quanto a compactação opcional iniciada durante a confirmação).
rocksdbWriterStallLatencyMs O tempo (em milissegundos) que o gravador esteve parado devido a uma compactação em segundo plano ou ao esvaziamento dos memtables para o disco.
rocksdbTotalBytesReadThroughIterator Algumas das operações com estado (como o processamento de tempo limite em flatMapGroupsWithState ou a marcação d'água em agregações em janela) exigem a leitura de todos os dados no banco de dados através de um iterador. O tamanho total dos dados não compactados lidos usando o iterador.