Tutoriel : Surveillez une requête de streaming structuré avec StreamingQueryListener

Un StreamingQueryListener reçoit les métriques de progression d’une requête Structured Streaming en cours d’exécution à la fin de chaque micro-lot, telles que le nombre de lignes en entrée, le débit de traitement et la durée du lot. Ce tutoriel vous guide pour attacher un écouteur qui vérifie si la requête suit le rythme des données en entrée, avertit quand la requête prend du retard, et enregistre la santé de chaque lot dans une table Unity Catalog que vous pouvez interroger avec SQL. L’exemple fonctionne sur le compute serverless et sur le compute compatible avec Unity Catalog.

Pour obtenir la référence complète des métriques et des champs d’événement signalés par un écouteur, consultez Surveillance des requêtes Structured Streaming sur Azure Databricks.

Ce tutoriel comprend les étapes suivantes :

1. Créer un carnet de notes
2. Créer les objets du catalogue Unity
3. Définir l’écouteur
4. Définir le processus d’écriture périodique
5. Démarrer la requête avec la surveillance activée
6. Interroger les données d’intégrité enregistrées
7. Arrêter la requête et nettoyer les ressources

Requirements

Pour suivre ce tutoriel, vous devez répondre aux exigences suivantes :

• Votre espace de travail doit avoir le catalogue Unity activé. Consultez Qu’est-ce que Unity Catalog ?.
• Vous devez avoir accès à un calcul serverless ou à un cluster compatible Unity Catalog. Pour les versions de Databricks Runtime et les modes d’accès que StreamingQueryListener prend en charge, voir Surveillance des requêtes de streaming structuré sur Azure Databricks.
• Vous devez avoir l’autorisation de créer un schéma, un volume et des tables dans un catalogue Unity Catalog. Ce tutoriel utilise un catalogue nommé main. Remplacez-le par un catalogue accessible en écriture.

Étape 1 : Créer un bloc-notes

Vous utilisez un notebook Databricks pour définir le listener et lancer la requête en streaming.

Pour créer un bloc-notes :

1. Cliquez sur Nouveau dans la barre latérale, puis sur Bloc-notes.
2. Dans le menu déroulant de calcul, sélectionnez calcul sans serveur ou votre cluster compatible Unity Catalog.
3. Sélectionnez Python comme langage par défaut.

Étape 2 : Créer les objets de Unity Catalog

L’écouteur enregistre les données d’intégrité de chaque lot dans une table managée Unity Catalog, la requête de streaming écrit dans une deuxième table managée et stocke son point de contrôle dans un volume Unity Catalog. Pour créer le schéma et le volume et définir les noms des objets, exécutez :

catalog = "main"
schema = "streaming_monitoring"

spark.sql(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{schema}")
spark.sql(f"CREATE VOLUME IF NOT EXISTS {catalog}.{schema}.checkpoints")

health_table = f"{catalog}.{schema}.streaming_health"
target_table = f"{catalog}.{schema}.rate_target"
checkpoint_path = f"/Volumes/{catalog}/{schema}/checkpoints/rate_target"

Étape 3 : Définir le listener

L’écouteur veille à ce que chaque fonction de rappel reste légère. Pour chaque micro-batch, il compare le débit de traitement à celui d’entrée, affiche un avertissement lorsque la requête prend du retard, et ajoute la santé du lot à un tampon en mémoire. Pour définir le listener, exécutez :

import threading
from pyspark.sql.streaming import StreamingQueryListener

class HealthListener(StreamingQueryListener):
    def __init__(self):
        self._lock = threading.Lock()
        self._buffer = []

    def onQueryStarted(self, event):
        print(f"Query started: {event.id}")

    def onQueryProgress(self, event):
        progress = event.progress
        input_rate = progress.inputRowsPerSecond
        processed_rate = progress.processedRowsPerSecond
        # The query is falling behind when it processes data slower than it arrives.
        falling_behind = input_rate > 0 and processed_rate < input_rate
        if falling_behind:
            print(f"WARNING batch {progress.batchId}: falling behind "
                  f"({processed_rate:.1f} processed/s < {input_rate:.1f} input/s)")
        with self._lock:
            self._buffer.append((int(progress.batchId), int(progress.numInputRows),
                                 float(input_rate), float(processed_rate),
                                 bool(falling_behind)))

    def drain(self):
        with self._lock:
            records, self._buffer = self._buffer, []
        return records

    def onQueryIdle(self, event):
        pass

    def onQueryTerminated(self, event):
        print(f"Query terminated: {event.id}")

L’écouteur redéfinit les méthodes d’événement suivantes :

La drain méthode retourne les enregistrements en mémoire tampon et efface le tampon. Un verrou protège le tampon car l’écouteur y ajoute depuis le thread d’écoute tandis que le processus d’écriture à l’étape suivante lit depuis un thread différent.

Étape 4 : Définir le processus d’écriture périodique

Un thread en arrière-plan écrit les enregistrements en mémoire tampon dans la table d’état de santé à un intervalle fixe, de sorte que le listener n’exécute jamais de job Spark sur son propre thread. La fonction publish est l’unique endroit où les enregistrements sont écrits. Pour envoyer les métriques vers un autre système, tel qu’Apache Kafka, remplacez le corps de publish. Pour définir le writer et sa boucle d’intervalles, exécutez :

def publish(records):
    if not records:
        return
    (spark.createDataFrame(records,
        "batchId long, numInputRows long, inputRowsPerSecond double, "
        "processedRowsPerSecond double, fallingBehind boolean")
     .write.mode("append").saveAsTable(health_table))

stop_event = threading.Event()

def flush_loop(listener, interval_seconds=10):
    while not stop_event.wait(interval_seconds):
        publish(listener.drain())

La boucle écrit un lot d’enregistrements toutes les 10 secondes jusqu’à ce que stop_event soit défini. Chaque écriture ajoute de nombreuses lignes à la fois, ce qui évite les petits fichiers qu’une écriture par lot créerait.

Étape 5 : Démarrer la requête avec la surveillance activée

Enregistrez l’écouteur, lancez le thread d’écriture, puis lancez une requête qui lit depuis la source rate intégrée et écrit dans la table cible. La source de débit génère des lignes à un rythme constant, ce qui permet à l’écouteur de signaler une progression régulière sans recourir à une source de données externe. Pour démarrer la surveillance et la requête, exécutez :

listener = HealthListener()
spark.streams.addListener(listener)

flusher = threading.Thread(target=flush_loop, args=(listener,), daemon=True)
flusher.start()

query = (
    spark.readStream.format("rate").option("rowsPerSecond", 5).load()
    .writeStream.format("delta")
    .option("checkpointLocation", checkpoint_path)
    .toTable(target_table)
)

Laissez la requête s’exécuter une ou deux minutes afin que plusieurs intervalles de flush s’écoulent. Chaque micro-lot déclenche onQueryProgress, qui affiche un avertissement lorsque la requête prend du retard et met en mémoire tampon les données d’intégrité du lot. L’auteur ajoute les enregistrements en mémoire tampon à la table toutes les 10 secondes, pour que vous puissiez les interroger pendant que le flux s’exécute.

Étape 6 : Interroger la santé enregistrée

Consultez la table d’intégrité pour connaître, pour chaque lot, le débit d’arrivée des données, le débit de traitement de la requête et déterminer si celle-ci a pris du retard. Pour afficher les enregistrements, exécutez :

display(spark.sql(f"""
  SELECT batchId, numInputRows,
         round(inputRowsPerSecond, 1) AS input_per_second,
         round(processedRowsPerSecond, 1) AS processed_per_second,
         fallingBehind
  FROM {health_table}
  ORDER BY batchId
"""))

La requête précédente retourne une ligne par lot complété, comme ceci :

batchId  numInputRows  input_per_second  processed_per_second  fallingBehind
0        2             0.0               1.9                   false
1        5             4.8               5.7                   false
2        5             5.7               4.6                   true
3        5             4.6               6.0                   false
4        4             4.8               5.0                   false
5        4             5.0               4.4                   true
6        5             5.5               6.1                   false
7        4             4.8               4.1                   true

Une requête en bon état traite les données au moins aussi vite qu’elles arrivent, donc processed_per_second reste à ou au-dessus de input_per_second et fallingBehind est false. Les lots individuels descendent souvent en dessous du débit d’entrée, comme le montrent les lots 2, 5 et 7. Ce tremblement est normal. Si fallingBehind reste true pendant plusieurs lots consécutifs, la requête ne parvient pas à suivre le rythme des données entrantes. Étudiez en ajoutant des ressources de calcul, en réduisant le travail dans chaque lot ou en diminuant le débit d’entrée.

Pour l’ensemble complet des métriques source, puits et état disponibles pour chaque événement de progression, voir Supervision des requêtes de streaming structuré sur Azure Databricks.

Étape 7 : Arrêter la requête et nettoyer les ressources

Arrêtez la requête, arrêtez le thread d’écriture, écrivez tous les enregistrements encore dans le tampon, et supprimez les objets que vous avez créés pour qu’ils ne continuent pas à générer des coûts. Pour nettoyer, exécutez :

query.stop()
stop_event.set()
flusher.join()
publish(listener.drain())
spark.streams.removeListener(listener)
spark.sql(f"DROP SCHEMA IF EXISTS {catalog}.{schema} CASCADE")

Ressources supplémentaires

• Surveillance des requêtes Structured Streaming sur Azure Databricks
• Concepts de streaming structuré
• Qu’est-ce que la diffusion en continu avec état ?
• Créer une application personnalisée avec état transformWithState