Tutorial: monitorar uma query do Structured Streaming com o StreamingQueryListener
O StreamingQueryListener recebe métricas de progresso de uma query em execução do Structured Streaming ao final de cada micro-lotes, como o número de linhas de entrada, a taxa de processamento e a duração do lote. Este tutorial o orienta a anexar um listener que verifica se a query mantém o ritmo de sua entrada, alerta quando a query está atrasada e registra a saúde de cada lote em uma tabela do Unity Catalog que você consulta com SQL. O exemplo é executado em compute serverless e em compute habilitado para Unity Catalog.
Para uma referência completa das métricas e dos campos de eventos que um listener relata, consulte Monitoramento de queries do Structured Streaming no Databricks.
Este tutorial inclui os seguintes passos:
- Criar um notebook
- Criar os objetos do Unity Catalog
- Definir o listener
- Definir o gravador periódico
- Começar a query com monitoramento
- Query a integridade registrada
- Parar a query e limpar
Requisitos
Para completar este tutorial, você deve atender aos seguintes requisitos:
- Seu workspace deve ter o Unity Catalog habilitado. Consulte O que é Unity Catalog?.
- Você deve ter acesso ao compute serverless ou a um cluster habilitado para Unity Catalog. Para as versões do Databricks Runtime e modos de acesso suportados pelo
StreamingQueryListener, consulte Monitoramento de queries do Structured Streaming no Databricks. - Você deve ter permissão para criar um esquema, um volume e tabelas em um catálogo do Unity Catalog. Este tutorial usa um catálogo chamado
main. Substitua por um catálogo no qual você pode gravar.
o passo 1: Criar um notebook
Você usa um notebook do Databricks para definir o listener e fazer a execução da query de transmissão.
Para criar um notebook:
- Clique em Novo na barra lateral e, em seguida, clique em
Notebook . - No menu suspenso de compute, selecione compute serverless ou seu cluster habilitado para Unity Catalog.
- Selecione Python como o idioma default.
o passo 2: Criar os objetos do Unity Catalog
O listener registra a integridade por lote em uma tabela gerenciada pelo Unity Catalog, a query de transmissão escreve em uma segunda tabela gerenciada e a query armazena seu checkpoint em um volume do Unity Catalog. Para criar o esquema e o volume e definir os nomes dos objetos, faça a execução:
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"
O passo 3: Definir o listener
O listener mantém cada callback leve. Para cada micro-lotes, ele compara a taxa de processamento com a taxa de entrada, imprime um aviso quando a query fica para trás e anexa o status dos lotes a um buffer na memória. Para definir o listener, execução:
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}")
O listener substitui os seguintes métodos de evento:
Método | Descrição |
|---|---|
| execução quando a query começar. |
| Executado ao final de cada micro-lotes. Este exemplo compara a taxa de processamento com a taxa de entrada, avisa quando a query está atrasada e armazena o status dos lotes em buffer. |
| Executado quando a query está parado e aguardando novos dados. |
| A execução ocorre quando a query para, com ou sem erro. |
O método drain retorna os registros em buffer e limpa o buffer. Um bloqueio protege o buffer porque o listener anexa a ele a partir da thread do listener enquanto o gravador na próxima etapa lê a partir de uma thread diferente.
onQueryProgress é uma execução na thread do listener, portanto, o trabalho pesado lá atrasa o relatório de progresso de cada query na sessão. É por isso que o listener só faz buffer no callback e escreve de uma thread separada. Consulte a nota em Monitoramento de queries do Structured Streaming no Databricks.
o passo 4: definir o gravador periódico
Uma thread em segundo plano grava os registros em buffer na tabela de integridade em um intervalo fixo, de modo que o listener nunca faz uma execução de um Job do Spark em sua própria thread. A função publish é o único local que grava os registros. Para enviar as métricas para outro sistema, como o Apache Kafka, substitua o corpo de publish. Para definir o gravador e seu loop de intervalo, execute:
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())
O loop escreve um lote de registros a cada 10 segundos até que stop_event seja definido. Cada gravação adiciona muitas linhas de uma vez, o que evita os arquivos pequenos que uma gravação por lote criaria.
O passo 5: Começar a query com monitoramento
Faça o registro do listener, comece o thread de gravação e, em seguida, comece uma query que lê da fonte de taxa integrada e grava na tabela de destino. A fonte de taxa gera linhas a uma taxa fixa, portanto o listener relata progresso constante sem uma fonte de dados externa. Para começar o monitoramento e a query, execução:
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)
)
Deixe a query em execução por um ou dois minutos para que vários intervalos de flush passem. Cada micro-lotes dispara onQueryProgress, que imprime um aviso quando a query está atrasada e armazena a saúde dos lotes. O writer anexa os registros armazenados à tabela a cada 10 segundos, para que você possa fazer query neles enquanto a transmissão é executada.
Passo 6: query a integridade registrada
Leia a tabela de saúde para ver, para cada lote, a rapidez com que os dados chegaram, a rapidez com que a query os processou e se a query ficou atrasada. Para exibir os registros, execução:
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
"""))
A query anterior retorna uma linha por lote concluído, como esta:
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
Uma query saudável processa os dados pelo menos tão rápido quanto eles chegam, de modo que processed_per_second permaneça em ou acima de input_per_second e fallingBehind seja false. Lotes individuais frequentemente caem abaixo da taxa de entrada, como os lotes 2, 5 e 7 mostram. Essa instabilidade é normal. Se fallingBehind permanecer true em vários lotes consecutivos, a query não consegue acompanhar sua entrada. Investigue adicionando compute, reduzindo o trabalho em cada lote ou diminuindo a taxa de entrada.
Para o conjunto completo de métricas de origem, destino e estado disponíveis em cada evento de progresso, consulte Monitoramento de queries do Structured Streaming no Databricks.
O passo 7: Parar a query e limpar
Pare a query, pare a thread de escrita, escreva todos os registros que ainda estão no buffer e descarte os objetos que você criou para que eles não continuem a gerar custos. Para limpar, execução:
query.stop()
stop_event.set()
flusher.join()
publish(listener.drain())
spark.streams.removeListener(listener)
spark.sql(f"DROP SCHEMA IF EXISTS {catalog}.{schema} CASCADE")