Tutorial: monitora una query di streaming strutturato con StreamingQueryListener

Un StreamingQueryListener listener riceve metriche di progresso da una query di Structured Streaming in esecuzione alla fine di ogni micro-batch, come il numero di righe di input, la velocità di elaborazione e la durata del batch. Questo tutorial ti guida nell'aggiunta di un listener che verifica se la query segue il suo input, avvisa quando la query è in ritardo e registra la salute di ogni batch in una tabella Unity Catalog che si interroga con SQL. L'esempio funziona sia su calcolo serverless che su calcolo abilitato a Unity Catalog.

Per un riferimento completo delle metriche e dei campi eventi che un ascoltatore segnala, consulta Monitoraggio delle query di streaming strutturato su Azure Databricks.

Questa esercitazione include i passaggi seguenti:

1. Crea un notebook
2. Crea gli oggetti del Catalogo Unity
3. Definisci il listener
4. Definisci il writer periodico
5. Inizia la query con il monitoraggio attivo
6. Consulta la salute registrata
7. Interrompi la query e pulisci

Requirements

Per completare questa esercitazione, è necessario soddisfare i requisiti seguenti:

• L'area di lavoro deve avere Unity Catalog abilitato. Consulta Che cos'è il Catalogo Unity?.
• Devi avere accesso a computazione serverless o a un cluster abilitato a Unity Catalog. Per le versioni di Databricks Runtime StreamingQueryListener e le modalità di accesso supportate, vedi Monitoraggio delle query di streaming strutturato su Azure Databricks.
• Devi disporre dell'autorizzazione per creare uno schema, un volume e delle tabelle in un catalogo di Unity Catalog. Questo tutorial utilizza un catalogo chiamato main. Sostituisci con un catalogo in cui puoi scrivere.

Passaggio 1: Creare un notebook

Usi un notebook Databricks per definire il listener ed eseguire la query in streaming.

Per creare un notebook:

1. Fare clic su Nuovo nella barra laterale, quindi fare clic Notebook.
2. Nel menu a tendina di calcolo, seleziona computazione serverless o il cluster abilitato al Unity Catalog.
3. Seleziona Python come linguaggio predefinito.

Passo 2: Crea gli oggetti del Catalogo Unity

L'ascoltatore registra la salute per batch su una tabella gestita Unity Catalog, la query streaming scrive in una seconda tabella gestita e la query memorizza il suo checkpoint in un volume Unity Catalog. Per creare lo schema e il volume e definire i nomi degli oggetti, esegui:

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"

Passo 3: Definisci il listener

L'ascoltatore mantiene ogni callback leggero. Per ogni micro-batch, confronta la velocità di elaborazione con quella di input, stampa un avviso quando la query è in ritardo e accoda lo stato di integrità del batch a un buffer in memoria. Per definire il listener, esegui:

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'ascoltatore ridefinisce i seguenti metodi di evento:

Metodo	Description
onQueryStarted	Si avvia quando inizia la query.
onQueryProgress	Viene eseguita alla fine di ogni micro-lotto. Questo esempio confronta la velocità di elaborazione con quella di input, avvisa quando la query è in ritardo e memorizza nel buffer lo stato di integrità del batch.
onQueryIdle	Si esegue quando la query è inattiva e in attesa di nuovi dati.
onQueryTerminated	Si esegue quando la query si interrompe, con o senza errore.

Il drain metodo restituisce i record bufferizzati e cancella il buffer. Un blocco protegge il buffer perché il listener vi aggiunge dal thread del listener mentre il writer nel passaggio successivo legge da un thread diverso.

Note

onQueryProgress gira sul thread listener, quindi il lavoro pesante lì ritarda la segnalazione dei progressi per ogni query della sessione. Per questo motivo l'ascoltatore buffera solo nel callback e scrive da un thread separato. Consulta la nota in Monitoring Structured Streaming queries su Azure Databricks.

Passo 4: Definisci il writer periodico

Un thread in background scrive i record bufferizzati nella tabella di salute a intervalli fissi, quindi l'ascoltatore non esegue mai un lavoro Spark sul proprio thread. La funzione publish è l'unico punto che scrive i record. Per inviare le metriche a un altro sistema, come Apache Kafka, si sostituisce il corpo di publish. Per definire il writer e il suo ciclo di intervallo, esegui:

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())

Il ciclo scrive un lotto di record ogni 10 secondi fino a quando stop_event non viene impostato. Ogni scrittura aggiunge molte righe alla volta, evitando così i piccoli file che una scrittura per batch creerebbe.

Passo 5: Avvia la query con il monitoraggio attivato

Registra il listener, avvia il thread del writer, poi avvia una query che legga dalla sorgente rate integrata e scriva nella tabella di destinazione. L'origine a velocità fissa genera righe a tasso fisso, così il listener segnala progressi costanti senza una fonte di dati esterna. Per iniziare a monitorare e a fare la query, esegui:

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

Lascia che la query venga eseguita per uno o due minuti in modo che passino diversi intervalli di flush. Ogni micro-batch attiva onQueryProgress, che stampa un avviso quando la query rimane indietro e memorizza lo stato di salute del batch. Il writer aggiunge i record bufferizzati alla tabella ogni 10 secondi, così puoi interrogarli mentre lo stream è in corso.

Passo 6: Consulta lo stato di integrità registrato

Leggi la tabella di stato per vedere, per ogni lotto, quanto velocemente arrivano i dati, quanto velocemente la query li processa e se la query è rimasta indietro. Per visualizzare i record, esegui:

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 precedente restituisce una riga per ogni batch completato, come questo:

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

Una query sana elabora i dati almeno con la stessa rapidità con cui arrivano, quindi processed_per_second rimane a o sopra input_per_second e fallingBehind è false. I singoli lotti spesso scendono sotto la velocità di ingresso, come mostrano i lotti 2, 5 e 7. Questo jitter è normale. Se fallingBehind rimane true per diversi lotti consecutivi, la query non riesce a tenere il passo con il suo input. Verificare aggiungendo risorse di calcolo, riducendo il lavoro in ogni batch o abbassando la velocità di input.

Per l'insieme completo delle metriche di origine, sink e stato disponibili per ogni evento di avanzamento, vedi Monitorare le query di Structured Streaming in Azure Databricks.

Passo 7: Ferma la query e ripulisci

Ferma la query, interrompi il thread di scrittura, scrivi tutti i record ancora nel buffer e elimina gli oggetti che hai creato in modo che non continuino a sostenere costi. Per pulire, esegui:

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

Risorse aggiuntive

• Monitoraggio delle query di streaming strutturato su Azure Databricks
• Concetti di Structured Streaming
• Che cos'è lo streaming con stato?
• Costruisci un'applicazione con stato personalizzata con transformWithState