Aller au contenu principal
Dernière mise à jour le

Optimiser et surveiller les performances des query en mode temps réel

Cette page couvre le réglage du compute, les techniques de réduction de la latence de bout en bout et les approches de mesure des performances de query en temps réel.

Réglage du compute​

Lorsque vous configurez votre compute, veuillez prendre en considération les points suivants :

  • Contrairement au mode micro-batch, les tâches en temps réel peuvent rester inactives en attendant les données, de sorte que le dimensionnement approprié est essentiel pour éviter le gaspillage de ressources.

  • Visez un niveau d'utilisation des clusters cible, tel que 50 %, en ajustant :

    • maxPartitions (pour Kafka)
    • spark.sql.shuffle.partitions (pour les étapes de shuffle)

  • Databricks recommande de configurer maxPartitions afin que chaque tâche gère plusieurs partitions Kafka pour réduire la surcharge.

  • Ajustez les emplacements de tâche par Worker pour qu'ils correspondent à la charge de travail pour les Job simples en une seule étape.

  • Pour les jobs à fort brassage, faites des essais pour trouver le nombre minimum de partitions de brassage qui évitent les retards et ajustez à partir de là. Le compute ne planifiera pas le job s'il n'a pas suffisamment d'emplacements.

remarque

À partir de Databricks Runtime 16.4 LTS et versions ultérieures, tous les pipelines en temps réel utilisent le checkpoint v2 pour permettre des transitions fluides entre les modes temps réel et micro-batch.

Optimisation de la latence​

Le mode temps réel de Structured Streaming propose des techniques facultatives pour réduire la latence de bout en bout. Aucun des deux n'est activé par default. Vous devez les activer séparément.

  • Suivi asynchrone de la progression: Déplace les écritures vers les Logs d'offset et de commit vers un thread asynchrone, réduisant le temps inter-batch pour les query sans état.
  • Point de contrôle d'état asynchrone: commence à traiter le micro-batch suivant dès que le calcul est terminé, sans attendre le point de contrôle d'état, réduisant ainsi la latence de la query avec état.

Monitoring et observabilité​

En mode temps réel, les métriques traditionnelles de durée de batch ne reflètent pas la latence réelle de bout en bout. Utilisez les approches ci-dessous pour mesurer la latence avec précision et identifier les goulets d'étranglement dans vos queries.

La latence de bout en bout est spécifique à la charge de travail et ne peut parfois être mesurée avec précision qu’à l’aide de la logique métier. Par exemple, si le Timestamp source est émis dans Kafka, vous pouvez calculer la latence comme la différence entre le Timestamp de sortie de Kafka et le Timestamp source.

Métriques intégrées avec StreamingQueryProgress​

L'événement StreamingQueryProgress est automatiquement enregistré dans les logs du Driver et accessible via la fonction de rappel onQueryProgress() de StreamingQueryListener. Cela vous permet de réagir aux événements de progression par programmation, par exemple, si vous souhaitez publier des métriques vers un système de monitoring externe. QueryProgressEvent.json() ou toString() incluent les métriques de mode en temps réel suivantes :

  1. **Latence de traitement**processingLatencyMs (). Le temps écoulé entre le moment où la query en mode temps réel lit un enregistrement et le moment où la query l'écrit à l'étape suivante ou en aval. Pour les requêtes à étape unique, cela mesure la même durée que la latence de bout en bout. Le système signale cette métrique par tâche.
  2. Latence de mise en file d'attente de la source (sourceQueuingLatencyMs). La durée écoulée entre le moment où le système écrit un enregistrement dans un bus de messages — par exemple, le temps d'ajout du log dans Kafka — et le moment où la query en mode temps réel lit l'enregistrement pour la première fois. Le système signale cette métrique par tâche.
  3. Latence de bout en bout (e2eLatencyMs). Le temps entre le moment où le système écrit l'enregistrement dans un bus de messages et le moment où la query en temps réel écrit l'enregistrement en aval. Le système agrège cette métrique par batch sur tous les enregistrements traités par toutes les tâches.

Par exemple :

JSON
"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
    }
}

Mesure de latence personnalisée avec l’API Observe​

L'API Observe vous permet de mesurer la latence en ligne sans lancer un Job distinct. Si vous disposez d'un timestamp source qui approche l'heure d'arrivée des données source, vous pouvez estimer la latence par batch en enregistrant un timestamp avant le sink et en calculant la différence. Les résultats apparaissent dans les rapports de progression et sont disponibles pour les auditeurs.

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.

Exemple de sortie :

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

Ressources supplémentaires​

Sur cette page