Ponto de verificação assíncrono de estado para consultas com estado

O ponto de verificação de estado assíncrono mantém garantias exatas uma vez para consultas de streaming, mas pode reduzir a latência geral para algumas cargas de trabalho com estado de Streaming Estruturado congestionadas em atualizações de estado. Isso é feito começando a processar o próximo microlote assim que o cálculo do microlote anterior tiver sido concluído sem esperar que o ponto de verificação de estado seja concluído. A tabela a seguir compara as compensações para pontos de verificação síncronos e assíncronos:

As seguintes são características de tarefas de streaming que podem beneficiar-se do ponto de verificação de estado assíncrono:

• O trabalho tem uma ou mais operações com estado (por exemplo, agregação, flatMapGroupsWithState, mapGroupsWithState, uniões de fluxo)

• A latência do ponto de verificação de estado é um dos principais contribuintes para a latência geral de execução em lote. Essas informações podem ser encontradas nos eventos StreamingQueryProgress. Esses eventos também são encontrados em logs do log4j no driver Spark. Aqui está um exemplo do progresso da consulta de streaming e como encontrar o impacto do ponto de verificação de estado na latência geral de execução em lote.

  • {
       "id" : "2e3495a2-de2c-4a6a-9a8e-f6d4c4796f19",
       "runId" : "e36e9d7e-d2b1-4a43-b0b3-e875e767e1fe",
       "...",
       "batchId" : 0,
       "durationMs" : {
         "...",
         "triggerExecution" : 547730,
         "..."
       },
       "stateOperators" : [ {
         "...",
         "commitTimeMs" : 3186626,
         "numShufflePartitions" : 64,
         "..."
       }]
    }
    

  • Análise da latência do ponto de verificação de estado do evento de progresso da consulta mencionada acima

    • A duração do lote (durationMs.triggerDuration) é de cerca de 547 segundos.
    • A latência de confirmação do armazenamento de estado (stateOperations[0].commitTimeMs) é de cerca de 3,186 segundos. A latência de confirmação é agregada nas tarefas que contêm um armazenamento de estado. Neste caso, existem 64 tarefas deste tipo (stateOperators[0].numShufflePartitions).
    • Cada tarefa contendo operador de estado levou uma média de 50 segundos (3.186/64) para o ponto de verificação. Esta é uma latência extra que se adiciona à duração do lote. Supondo que todas as 64 tarefas estejam sendo executadas simultaneamente, a etapa do ponto de verificação contribuiu com cerca de 9% (50 segundos / 547 segundos) da duração do lote. A porcentagem fica ainda maior quando o máximo de tarefas simultâneas é inferior a 64.

Ativação do ponto de verificação de estado assíncrono

Você deve usar o armazenamento de estado baseado em RocksDB para pontos de verificação de estado assíncrono. Defina as seguintes configurações:

spark.conf.set(
  "spark.databricks.streaming.statefulOperator.asyncCheckpoint.enabled",
  "true"
)

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

Limitações e requisitos para pontos de verificação assíncronos

• Qualquer falha em um ponto de verificação assíncrono em qualquer um ou mais armazenamentos falhará na consulta. No modo de ponto de verificação síncrono, o ponto de verificação é executado como parte da tarefa e o Spark tenta novamente a tarefa várias vezes antes de falhar na consulta. Este mecanismo não está presente com a verificação de estado assíncrona. O Databricks recomenda o uso de trabalhos contínuos para repetições automáticas em caso de falha de trabalho. Veja Executar tarefas continuamente.
• O ponto de verificação assíncrono funciona melhor quando os locais de armazenagem de estado não são alterados entre as execuções de micro-lotes. O redimensionamento do cluster, em combinação com o ponto de verificação de estado assíncrono, pode não funcionar bem porque a instância de armazenamentos de estado pode ser redistribuída à medida que os nós são adicionados ou excluídos como parte do evento de redimensionamento do cluster.
• O ponto de verificação de estado assíncrono é suportado apenas na implementação do provedor de armazenamento de estado RocksDB. A implementação de armazenamento de estado na memória padrão não oferece suporte a ele.