Tutoriel : Surveiller une query Structured Streaming avec StreamingQueryListener
Un StreamingQueryListener reçoit des mesures de progression d’une query Structured Streaming en cours d’exécution à la fin de chaque micro-batch, telles que le nombre de lignes d’entrée, la vitesse de traitement et la durée du batch. Ce tutoriel vous guide tout au long de l’attachement d’un écouteur qui vérifie si la query suit son entrée, avertit lorsque la query prend du retard et enregistre l’état de chaque batch dans une table Unity Catalog que vous interrogez avec SQL. L’exemple s’exécute sur un compute serverless et sur un compute compatible avec Unity Catalog.
Pour une référence complète des mesures et des champs d’événement rapportés par un écouteur, voir monitoring des query Structured Streaming sur Databricks.
Ce tutoriel comprend les étapes suivantes :
- Créer un notebook
- Créer les objets Unity Catalog
- Définir l’écouteur
- Définir l’écrivain périodique
- start la query avec monitoring
- query l’état enregistré
- Arrêtez la query et effectuez le nettoyage
Conditions requises
Pour suivre ce tutoriel, vous devez remplir les conditions suivantes :
- Votre workspace doit avoir Unity Catalog activé. Voir Qu’est-ce que Unity Catalog ?.
- Vous devez avoir accès au serverless compute ou à un cluster avec Unity Catalog activé. Pour les versions de Databricks Runtime et les modes d'accès pris en charge par
StreamingQueryListener, consultez monitoring des query Structured Streaming sur Databricks. - Vous devez disposer de l’autorisation de créer un schéma, un volume et des tables dans un catalogue Unity Catalog. Ce didacticiel utilise un catalogue nommé
main. Remplacez par un catalogue dans lequel vous pouvez écrire.
Étape 1 : Créer un notebook
Vous utilisez un Notebook Databricks pour définir l’écouteur et exécuter la query de streaming.
Pour créer un notebook :
- Cliquez sur Nouveau dans la barre latérale, puis cliquez sur
Notebook . - Dans le menu déroulant compute, sélectionnez serverless compute ou votre cluster compatible Unity Catalog.
- Sélectionnez Python comme langage default.
Étape 2 : Créer les objets Unity Catalog
L’écouteur enregistre l’état de chaque batch dans une table gérée par Unity Catalog, la query de streaming écrit dans une deuxième table gérée et la query 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 d’objet, 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 l’écouteur
L’écouteur permet de garder chaque rappel léger. Pour chaque micro-batch, il compare le taux de traitement au taux d’entrée, affiche un avertissement lorsque la query est en retard et ajoute l’état de santé du batch à une mémoire tampon. Pour définir l’écouteur, 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 remplace les méthodes d’événement suivantes :
Méthode | Description |
|---|---|
| S’exécute au start de la query. |
| S'exécute à la fin de chaque micro-batch. Cet exemple compare le taux de traitement au taux d’entrée, avertit lorsque la query prend du retard et met en mémoire tampon l’état de santé du batch. |
| S’exécute lorsque la query est inactive et en attente de nouvelles données. |
| S’exécute lorsque la query s’arrête, avec ou sans erreur. |
La méthode drain retourne les enregistrements mis en tampon et efface le tampon. Un verrou protège le tampon, car l’écouteur y ajoute des éléments à partir du thread de l’écouteur tandis que l’écrivain à l’étape suivante lit à partir d’un thread différent.
onQueryProgress s’exécute sur le thread de l’écouteur ; par conséquent, un travail lourd à cet endroit retarde le rapport de progression pour chaque query de la session. C’est pourquoi l’écouteur ne met en mémoire tampon que dans le rappel et écrit à partir d’un thread distinct. Consultez la note dans monitoring des query Structured Streaming sur Databricks.
Étape 4 : Définir l’écrivain périodique
Un thread en arrière-plan écrit les enregistrements mis en tampon dans la table d’intégrité à un intervalle fixe, de sorte que l’écouteur n’exécute jamais de job Spark sur son propre thread. La fonction publish est le seul emplacement qui écrit les enregistrements. Pour envoyer les mesures à un autre système, comme Apache Kafka, remplacez le corps de publish. Pour définir l’auteur et sa boucle d’intervalle, 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 batch d'enregistrements toutes les 10 secondes jusqu'à ce que stop_event soit défini. Chaque écriture ajoute plusieurs lignes à la fois, ce qui évite les petits fichiers qu'une écriture par batch créerait.
Étape 5 : Start la query avec le monitoring
Enregistrez le listener, start le thread d'écriture, puis start une query qui lit à partir de la source de taux intégrée et écrit dans la table cible. La source de taux génère des lignes à un taux fixe, de sorte que le listener signale une progression constante sans source de données externe. Pour start la monitoring et la query, 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 query s'exécuter pendant une ou deux minutes pour que plusieurs intervalles de vidage s'écoulent. Chaque micro-batch Trigger,onQueryProgress qui affiche un avertissement lorsque la query est en retard et met en tampon l'état de santé du batch. L'écrivain ajoute les enregistrements mis en tampon à la table toutes les 10 secondes, vous pouvez donc les query pendant l'exécution du Stream.
Étape 6 : query l’état enregistré
Lisez la table de santé pour voir, pour chaque batch, la rapidité d'arrivée des données, la rapidité de traitement par la query et si la query 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 query précédente retourne une ligne par batch terminé, comme celle-ci :
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 query saine traite les données au moins aussi rapidement qu’elles arrivent ; processed_per_second reste donc au-dessus de input_per_second ou égal à celui-ci, et fallingBehind est false. Les batchs individuels tombent souvent en dessous du taux d’entrée, comme le montrent les batchs 2, 5 et 7. Ce jitter est normal. Si fallingBehind reste true sur plusieurs batch consécutifs, la query ne peut pas suivre son entrée. Menez une enquête en ajoutant du compute, en réduisant la charge de travail de chaque batch ou en diminuant le taux d’entrée.
Pour obtenir l’ensemble complet des métriques de source, de sink et d’état disponibles sur chaque événement de progression, consultez monitoring des queries Structured Streaming sur Databricks.
Étape 7 : Arrêtez la query et effectuez le nettoyage
Arrêtez la query, arrêtez le thread d’écriture, écrivez les enregistrements encore présents dans le tampon et supprimez les objets que vous avez créés afin 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")