Optimización y supervisión del rendimiento de las consultas en modo en tiempo real

En esta página se tratan el ajuste de proceso, las técnicas para reducir la latencia de un extremo a otro y los enfoques para medir el rendimiento de las consultas en modo en tiempo real.

Ajuste del cálculo

Al configurar su computación, tenga en cuenta lo siguiente:

• A diferencia del modo microlote, las tareas en tiempo real pueden permanecer inactivas mientras esperan datos, por lo que el ajuste de tamaño correcto es esencial para evitar los recursos desperdiciados.
• El objetivo es alcanzar un nivel de utilización del clúster objetivo del 50%, ajustando:
  • maxPartitions (para Kafka)
  • spark.sql.shuffle.partitions (para etapas aleatorias)
• Databricks recomienda establecer maxPartitions para que cada tarea controle varias particiones de Kafka para reducir la sobrecarga.
• Ajuste las ranuras de tareas por trabajador para que se ajusten a la carga de trabajo en trabajos sencillos de una sola fase.
• Para trabajos con mucha aleatoriedad, pruebe hasta encontrar el número mínimo de particiones aleatorias que evite los retrasos y ajuste a partir de ahí. El sistema de proceso no programará el trabajo si no dispone de suficientes ranuras.

Optimización de latencia

El modo en tiempo real de Structured Streaming tiene técnicas opcionales para reducir la latencia de un extremo a otro. Ninguno está habilitado de forma predeterminada. Debe habilitarlos por separado.

• Seguimiento asíncrono del progreso: traslada las escrituras en los registros de desplazamiento y de confirmación a un hilo asíncrono, lo que reduce el tiempo entre lotes para las consultas sin estado.
• Puntos de control de estado asíncronos: Comienza a procesar el siguiente microlote tan pronto como finaliza el cálculo, sin esperar a que se realice el punto de control de estado, lo que reduce la latencia en las consultas con estado.

Supervisión y observabilidad

En el modo en tiempo real, las métricas tradicionales de duración por lotes no reflejan la latencia real de un extremo a otro. Use los enfoques siguientes para medir la latencia con precisión e identificar cuellos de botella en las consultas.

La latencia de un extremo a otro es específica de la carga de trabajo y a veces solo se puede medir con precisión con la lógica de negocios. Por ejemplo, si la marca de tiempo de origen se genera en Kafka, puede calcular la latencia como la diferencia entre la marca de tiempo generada por Kafka y la marca de tiempo de origen.

Métricas integradas con StreamingQueryProgress

El evento StreamingQueryProgress se registra automáticamente en los registros del controlador y se puede acceder a él a través de la función de devolución de llamada onQueryProgress() de StreamingQueryListener. Esto le permite reaccionar a los eventos de progreso mediante programación, por ejemplo, si desea publicar métricas en un sistema de supervisión externo. QueryProgressEvent.json() o toString() incluyen estas métricas en modo de tiempo real:

1. Latencia de procesamiento (processingLatencyMs). El tiempo transcurrido entre el momento en que la consulta en modo en tiempo real lee un registro y el momento en que la consulta lo escribe en la siguiente etapa o en el destino. Para las consultas de una sola fase, esto mide la misma duración que la latencia de un extremo a otro. El sistema notifica esta métrica por tarea.
2. Latencia de puesta en cola de origen (sourceQueuingLatencyMs). La cantidad de tiempo transcurrido entre cuando el sistema escribe un registro en un bus de mensajes (por ejemplo, el tiempo de anexión del registro en Kafka) y cuando la consulta en modo en tiempo real lee primero el registro. El sistema notifica esta métrica por tarea.
3. Latencia de un extremo a otro (e2eLatencyMs). Tiempo entre el momento en que el sistema escribe el registro en un bus de mensajes y el momento en que la consulta en modo en tiempo real escribe el registro aguas abajo. El sistema agrega esta métrica por lote en todos los registros procesados por todas las tareas.

Por ejemplo:

"rtmMetrics" : {
    "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
    }
}

Medición de latencia personalizada con observe API

La API de Observación le permite medir la latencia en línea sin iniciar un trabajo independiente. Si dispones de una marca de tiempo de origen que se aproxima a la hora de llegada de los datos de origen, puedes estimar la latencia por lote registrando una marca de tiempo antes del destino y calculando la diferencia. Los resultados aparecen en los informes de progreso y están disponibles para los oyentes.

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.

Scala

import org.apache.spark.sql.functions.{avg, col, lit, max, percentile_approx, udf, unix_millis}

val currentTimestampUDF = udf(() => System.currentTimeMillis())

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

Resultados del ejemplo:

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

Recursos adicionales

• Referencia del modo en tiempo real