Pular para o conteúdo principal
Última atualização em

Otimize e monitore o desempenho das consultas em modo de tempo real.

Esta página aborda otimização compute , técnicas para reduzir a latência de ponta a ponta e abordagens para medir o desempenho de consultas no modo tempo real.

ajuste computacional​

Ao configurar seu compute, considere o seguinte:

  • Ao contrário do modo micro-lotes, o modo tempo real tarefa pode permanecer parado enquanto se aguarda dados, portanto, o dimensionamento correto é essencial para evitar o desperdício de recursos.

  • Defina uma meta de utilização do cluster, como 50%, ajustando as seguintes configurações:

    • maxPartitions (para Kafka)
    • spark.sql.shuffle.partitions (para fases aleatórias)

  • A Databricks recomenda definir maxPartitions para que cada tarefa lide com várias partições do Kafka para reduzir a sobrecarga.

  • Ajuste os slots de tarefas por worker para corresponder à carga de trabalho de um Job simples de uma única etapa.

  • Para trabalhos com muitas operações de embaralhamento (shuffle), experimente para encontrar o número mínimo de partições de embaralhamento que evitam acúmulos e ajuste a partir daí. O compute não irá programar a tarefa se não tiver slots suficientes.

nota

A partir do Databricks Runtime 16.4 LTS e versões superiores, todos os pipelines Tempo Real utilizam o checkpoint v2 para permitir transições perfeitas entre os modos Tempo Real e Micro-Lotes.

Otimização de latência​

O modo de transmissão estruturada tempo real possui técnicas opcionais para redução da latência ponta a ponta. Nenhuma das duas está ativada por default. Você precisa habilitá-los separadamente.

  • Acompanhamento de progresso assíncrono: Move as gravações nos logs de deslocamento e commit para uma thread assíncrona, reduzindo o tempo de interpolação para consultas sem estado.
  • Checkpoint de estado assíncrono: Inicia o processamento dos próximos microlotes assim que a computação for concluída, sem esperar pelo checkpoint de estado, reduzindo a latência para consultas com estado.

monitoramento e observabilidade​

No modo tempo real, os lotes de medição tradicionais não refletem a latência real de ponta a ponta. Utilize as abordagens abaixo para medir a latência com precisão e identificar gargalos em suas consultas.

A latência de ponta a ponta é específica para cada carga de trabalho e, às vezes, só pode ser medida com precisão por meio da lógica de negócios. Por exemplo, se o timestamp de origem for enviado para o Kafka, você pode calcular a latência como a diferença entre o timestamp de saída do Kafka e o timestamp de origem.

redes integradas com StreamingQueryProgress​

O evento StreamingQueryProgress é registrado automaticamente nos logs do driver e acessível através da função de retorno de chamada onQueryProgress() de StreamingQueryListener. Isso permite que você reaja a eventos de progresso de forma programática, por exemplo, se quiser publicar métricas em um sistema de monitoramento externo. QueryProgressEvent.json() ou toString() incluem estas métricas de modo em tempo real:

  1. Processing latency (processingLatencyMs). The time elapsed between when the tempo real mode query reads a record and when the query writes it to the next stage or downstream. O sistema relata essa métrica por tarefa.
  2. Latência de enfileiramento de origem (sourceQueuingLatencyMs). O tempo decorrido entre o momento em que o sistema grava um registro em um barramento de mensagens — por exemplo, o tempo de acréscimo de log no Kafka — e o momento em que a consulta no modo em tempo real lê o registro pela primeira vez. O sistema reporta essas métricas por tarefa.
  3. Latência de ponta a ponta (e2eLatencyMs). O intervalo de tempo entre o momento em que o sistema grava o registro em um barramento de mensagens e o momento em que a consulta em modo de tempo real grava o registro posteriormente. O sistema agrega essas métricas por lote em todos os registros processados por todas as tarefas.

The JSON progress event includes these métricas under latencies. Por exemplo:

JSON
{
  "latencies": {
    "processingLatencyMs": {
      "P0": 0,
      "P50": 0,
      "P90": 0,
      "P95": 0,
      "P99": 0
    },
    "sourceQueuingLatencyMs": {
      "P0": 0,
      "P50": 1,
      "P90": 1,
      "P95": 2,
      "P99": 3
    },
    "e2eLatencyMs": {
      "P0": 0,
      "P50": 1,
      "P90": 1,
      "P95": 2,
      "P99": 4
    }
  }
}

Monitorização da utilização da tarefa​

Use busyTimeFraction to determine whether Spark tarefa utilization is limiting throughput. Values near 1 indicate that tarefas are fully utilized and the query might need more compute. Lower values indicate that tarefas spend more time parado or blocked. The metric ranges from 0 to 1 and is reported per stage and tarefa.

Enable debug métricas before starting the transmissão query:

Python
spark.conf.set("spark.databricks.streaming.execution.enableDebugMetrics", "true")

Use one of the following methods to access tarefa métricas:

Começar or restart the query. After a Trigger completes, open dados brutos abaixo the query cell and find _taskMetrics under latencies. O exemplo a seguir mostra as métricas da tarefa:

JSON
{
  "latencies": {
    "_taskMetrics": {
      "stage_0_task_0": {
        "busyTimeFraction": 0.03
      }
    }
  }
}

Medição de latência personalizada com a API Observe​

A API Observe permite medir a latência em linha, sem a necessidade de executar uma tarefa separada. Se você tiver um registro de data e hora da origem que se aproxime do horário de chegada dos dados da origem, poderá estimar a latência por lote registrando um registro de data e hora antes do destino e calculando a diferença. Os resultados aparecem em relatórios de progresso e ficam disponíveis para os ouvintes.

Python
from datetime import datetime

from pyspark.sql.functions import avg, col, lit, max, percentile_approx, udf, unix_millis
from pyspark.sql.types import TimestampType

@udf(returnType=TimestampType())
def current_timestamp():
  return datetime.now()

# Query before outputting
.withColumn("temp-timestamp", current_timestamp())
.withColumn(
  "latency",
  unix_millis(col("temp-timestamp")).cast("long") - unix_millis(col("timestamp")).cast("long"))
.observe(
  "observedLatency",
  avg(col("latency")).alias("avg"),
  max(col("latency")).alias("max"),
  percentile_approx(col("latency"), lit(0.99), lit(150)).alias("p99"),
  percentile_approx(col("latency"), lit(0.5), lit(150)).alias("p50"))
.drop(col("latency"))
.drop(col("temp-timestamp"))
# Output part of the query. For example, .WriteStream, etc.

Exemplo de saída:

JSON
"observedMetrics" : {
  "observedLatency" : {
    "avg" : 63.8369765176552,
    "max" : 219,
    "p99" : 154,
    "p50" : 49
  }
}

Recursos adicionais​

Nesta página