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) : durée écoulée entre le moment où la query en mode temps réel lit un enregistrement pour la première fois et celui où elle start à l'écrire dans le sink. Cela inclut toutes les étapes de traitement mais exclut le temps passé à écrire dans le sink.
  2. Latence de mise en file d'attente source (sourceQueuingLatencyMs) : durée entre le moment où le système écrit un enregistrement dans un bus de messages (comme le temps d'ajout de Logs dans Kafka) et le moment où la query en mode temps réel lit l'enregistrement pour la première fois.
  3. Latence de bout en bout (e2eLatencyMs) : durée entre l'écriture d'un enregistrement par le système dans un bus de messages et le start de son écriture par la query en mode temps réel dans la destination. Cela exclut le temps passé à l'écriture dans la destination.

Spark agrège ces métriques par batch pour tous les enregistrements traités et les signale sous forme de distributions de centiles. L'événement de progression JSON inclut ces métriques sous latencies. Par exemple :

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

Surveiller l’utilisation des tâches​

Utilisez busyTimeFraction pour déterminer si l'utilisation des tâches Spark limite le throughput. Des valeurs proches de 1 indiquent que les tâches sont pleinement utilisées et que la query peut avoir besoin de plus de compute. Des valeurs plus faibles indiquent que les tâches passent plus de temps inactives ou bloquées. La métrique est comprise entre 0 et 1 et est signalée par étape et par tâche.

Activez les métriques de débogage avant de démarrer la query de streaming :

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

Utilisez l’une des méthodes suivantes pour accéder aux métriques de tâche :

start ou redémarrer la query. Une fois qu'un trigger est terminé, ouvrez Raw Data sous la cellule de query et recherchez _taskMetrics sous latencies. L'exemple suivant présente les indicateurs de tâche :

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

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