Como aprendi Apache Spark — Parte 15: Structured Streaming, Watermarks e Checkpoints
Nota histórica: esta série revisita minha trajetória em Engenharia de Dados iniciada em 2012. O Structured Streaming foi introduzido posteriormente, na evolução do Apache Spark 2.x. Este capítulo conecta os fundamentos do processamento distribuído às arquiteturas modernas de streaming.
Nos capítulos anteriores, estudamos como o Apache Spark distribui processamento, organiza Tasks, gerencia memória, executa shuffle e otimiza consultas.
Também exploramos formatos de armazenamento, arquiteturas Lakehouse, qualidade de dados e governança.
Agora vamos enfrentar um problema diferente.
Até aqui, muitos exemplos partiam de um dataset finito.
Líamos os dados, aplicávamos transformações e gravávamos o resultado.
Mas e quando os dados não param de chegar?
Imagine uma plataforma que recebe continuamente:
- Transações financeiras.
- Pedidos de e-commerce.
- Eventos de aplicativos.
- Telemetria de equipamentos.
- Atualizações de sistemas operacionais.
- Mensagens de dispositivos IoT.
Não queremos necessariamente esperar até o dia seguinte para processar tudo.
Queremos transformar eventos em informações úteis conforme eles chegam.
É nesse contexto que entra o Apache Spark Structured Streaming.
A pergunta central deste capítulo é:
Como processar dados potencialmente infinitos utilizando um mecanismo distribuído, com controle de estado, recuperação de falhas e semântica temporal?
1. Batch vs. Streaming
No processamento batch, trabalhamos normalmente com um conjunto finito de dados.
BATCH PROCESSING
Data Source
|
v
Finite Dataset
|
v
Spark Job
|
v
Result
Por exemplo:
df = spark.read.parquet(
"/data/vendas/2026-11-05"
)
resultado = (
df.groupBy("categoria")
.count()
)
resultado.write.parquet(
"/data/resultados"
)
A aplicação processa os dados disponíveis e termina.
No streaming, a fonte pode continuar produzindo eventos indefinidamente.
STREAM PROCESSING
Event 1 ----+
Event 2 ----+
Event 3 ----+----> Spark Streaming Query
Event 4 ----+ |
Event 5 ----+ v
... Continuous Results
O processamento acompanha a evolução da fonte.
Isso não significa que cada evento seja necessariamente processado individualmente e imediatamente.
A estratégia de execução depende do mecanismo utilizado.
2. O que é Structured Streaming?
Structured Streaming é a API de processamento de fluxos do Apache Spark baseada no modelo estruturado de DataFrames e Spark SQL.
Sua principal proposta é permitir que desenvolvedores utilizem operações familiares para trabalhar com dados que chegam continuamente.
Em batch:
df = spark.read.parquet(
"/data/eventos"
)
Em streaming:
df = (
spark.readStream
.format("parquet")
.load("/data/eventos")
)
A diferença parece pequena.
Mas o modelo de execução muda.
O DataFrame streaming representa uma consulta sobre uma fonte que pode receber novos dados.
O Spark passa a executar essa consulta incrementalmente, conforme a fonte disponibiliza informações.
Nem todas as operações permitidas em DataFrames batch são suportadas em consultas streaming, e algumas exigem modos de saída ou configurações específicos.
3. O modelo de tabela ilimitada
Uma forma útil de entender Structured Streaming é imaginar um fluxo como uma tabela que cresce continuamente.
TIME T1
+----------+--------+
| event_id | value |
+----------+--------+
| 1 | 100 |
+----------+--------+
TIME T2
+----------+--------+
| event_id | value |
+----------+--------+
| 1 | 100 |
| 2 | 200 |
+----------+--------+
TIME T3
+----------+--------+
| event_id | value |
+----------+--------+
| 1 | 100 |
| 2 | 200 |
| 3 | 300 |
+----------+--------+
O Spark mantém uma consulta lógica sobre essa tabela.
À medida que novos dados chegam, o mecanismo atualiza o resultado de acordo com a operação e o modo de execução.
Esse modelo ajuda a conectar streaming ao conhecimento de Spark SQL que já construímos.
Mas existe um desafio.
Se a tabela nunca para de crescer, não podemos simplesmente recalcular todo o histórico a cada evento.
Precisamos de execução incremental.
4. Microbatch: o modelo mais utilizado
O mecanismo padrão do Structured Streaming é o processamento por microbatches.
Conceitualmente:
EVENT STREAM
|
v
+------------+
| Microbatch |
| 1 |
+------------+
|
v
Spark Execution
+------------+
| Microbatch |
| 2 |
+------------+
|
v
Spark Execution
+------------+
| Microbatch |
| 3 |
+------------+
|
v
Spark Execution
Cada microbatch processa um conjunto de dados novos identificado pelo mecanismo de streaming.
O Spark reutiliza componentes de seu motor distribuído para executar as transformações.
Isso permite aproveitar conceitos já conhecidos:
- Driver.
- Executors.
- Tasks.
- Partições.
- Shuffle.
- Spark SQL.
- Planejamento físico.
Structured Streaming não elimina esses fundamentos.
Ele os utiliza em um modelo de execução contínua.
5. Microbatch não significa latência zero
Imagine uma consulta configurada para iniciar novos microbatches aproximadamente a cada cinco segundos.
.trigger(processingTime="5 seconds")
Isso não garante que todos os eventos serão processados em exatamente cinco segundos.
A latência depende de:
- Disponibilidade dos dados.
- Tempo de processamento.
- Scheduling.
- Capacidade do cluster.
- Shuffle.
- Escrita no destino.
- Backpressure.
- Comportamento da fonte.
Se um microbatch demora mais do que o intervalo configurado, a consulta pode acumular atraso.
Portanto, precisamos distinguir:
Trigger Interval
Frequência desejada para verificar e iniciar processamento.
Processing Time
Tempo necessário para executar um microbatch.
End-to-End Latency
Tempo entre a ocorrência do evento e sua disponibilidade no destino.
Essas métricas não são equivalentes.
6. Sources e Sinks
Uma consulta Structured Streaming possui três componentes conceituais:
SOURCE
|
v
TRANSFORMATION
|
v
SINK
A Source fornece os dados.
Exemplos:
- Kafka.
- Arquivos.
- Rate source para testes.
- Outras fontes compatíveis.
As Transformations representam as operações Spark.
Exemplos:
- Filter.
- Select.
- Aggregation.
- Join.
- Window.
O Sink recebe o resultado.
Exemplos:
- Console.
- Arquivos.
- Tabelas compatíveis.
- Kafka.
-
foreachBatch.
As combinações permitidas dependem das operações, do modo de saída e das capacidades de cada conector.
7. Primeiro laboratório: streaming sem Kafka
Vamos começar com um exemplo executável localmente.
Utilizaremos a fonte rate, que gera linhas contendo um timestamp e um valor numérico.
Ela é útil para aprendizado e testes.
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.master("local[2]")
.appName("StructuredStreamingLab")
.getOrCreate()
)
eventos = (
spark.readStream
.format("rate")
.option("rowsPerSecond", 5)
.load()
)
Essa fonte disponibiliza duas colunas principais:
timestamp
value
Agora podemos transformar os dados:
from pyspark.sql import functions as F
eventos_transformados = (
eventos
.withColumn(
"categoria",
F.when(
F.col("value") % 2 == 0,
F.lit("PAR")
).otherwise(
F.lit("IMPAR")
)
)
)
Ainda não iniciamos a execução.
Assim como nos DataFrames batch, definimos um plano lógico de transformação.
8. Iniciando uma streaming query
Para executar o processamento, utilizamos writeStream.
query = (
eventos_transformados
.writeStream
.format("console")
.outputMode("append")
.option("truncate", False)
.trigger(processingTime="5 seconds")
.start()
)
O Spark inicia uma consulta streaming.
O console passa a apresentar novos registros processados em microbatches.
Para encerrar:
query.stop()
Em uma aplicação de longa duração, podemos utilizar:
query.awaitTermination()
Esse método bloqueia a execução aguardando o encerramento da consulta.
No laboratório, execute stop() em outra célula ou utilize um encerramento programado para não manter o processo indefinidamente.
9. O papel dos Output Modes
Structured Streaming oferece modos de saída para determinadas consultas.
Os três principais são:
Append
Emite novas linhas de resultado quando elas podem ser consideradas finais para a operação suportada.
Update
Emite linhas que foram atualizadas desde o último trigger, conforme a consulta e o sink.
Complete
Emite a tabela de resultados completa a cada trigger.
A disponibilidade desses modos depende do tipo de operação.
Por exemplo, uma agregação streaming sem watermark pode precisar manter resultados atualizáveis continuamente.
Nesse cenário, append não é automaticamente apropriado.
É importante validar a compatibilidade do modo com a consulta.
10. Event Time vs. Processing Time
Agora chegamos a um dos conceitos mais importantes de streaming.
Imagine um evento que ocorreu às 10h00.
Mas chegou ao Spark às 10h07.
Temos dois tempos diferentes.
Event Time
O momento em que o evento aconteceu.
Processing Time
O momento em que o sistema o processou.
Conceitualmente:
EVENT OCCURS
10:00
|
|
| Network / Queue / Delay
|
v
EVENT PROCESSED
10:07
Essa diferença é fundamental para análises temporais.
Imagine calcular o número de transações por minuto.
Devemos agrupar pela hora em que a transação ocorreu ou pela hora em que foi processada?
A resposta depende do requisito de negócio.
Para muitas análises de eventos, queremos utilizar Event Time.
11. Eventos fora de ordem
Em sistemas distribuídos, eventos nem sempre chegam na mesma ordem em que ocorreram.
Imagine:
Arrival Order:
Event A → 10:05
Event B → 10:03
Event C → 10:04
O evento B chegou depois de A, embora tenha ocorrido antes.
Isso pode acontecer por:
- Latência de rede.
- Retries.
- Buffers.
- Partições de mensageria.
- Falhas temporárias.
- Processamento assíncrono.
Se a aplicação assumir que os eventos chegam ordenados, os resultados podem ficar incorretos.
Precisamos de uma estratégia para lidar com atrasos e desordenação.
12. Window Aggregations
Structured Streaming permite realizar agregações por janelas temporais.
Imagine contar eventos em intervalos de um minuto.
contagem_por_janela = (
eventos
.groupBy(
F.window(
F.col("timestamp"),
"1 minute"
)
)
.count()
)
Essa consulta cria grupos baseados no timestamp dos eventos.
Conceitualmente:
10:00 ───────── 10:01
Window 1
10:01 ───────── 10:02
Window 2
10:02 ───────── 10:03
Window 3
Os eventos são associados às janelas correspondentes.
Mas surge uma pergunta.
Quando podemos considerar que uma janela está completa?
Se eventos atrasados podem chegar indefinidamente, o resultado também pode continuar mudando.
É aqui que entram Watermarks.
13. O que é Watermark?
Uma Watermark é um mecanismo utilizado pelo Spark para acompanhar o progresso do tempo dos eventos e estabelecer limites para o tratamento de dados atrasados em operações compatíveis.
Exemplo:
eventos_com_watermark = (
eventos
.withWatermark(
"timestamp",
"10 minutes"
)
)
Isso não significa simplesmente:
Descartar todo evento que chegou mais de dez minutos depois de acontecer.
A semântica é mais específica.
O mecanismo acompanha o maior Event Time observado e utiliza o atraso configurado para determinar o avanço da watermark.
De forma conceitual, em uma consulta com uma única watermark:
Watermark ≈
Maximum Observed Event Time
-
Allowed Delay
O comportamento exato depende da operação, da execução e da política de watermark utilizada.
14. Como a Watermark afeta as janelas?
Imagine uma janela:
10:00 até 10:05
E uma watermark que já avançou para além do encerramento dessa janela.
Em uma agregação por janela compatível, o Spark pode considerar que não precisa mais manter aquele estado.
Isso permite liberar recursos.
Conceitualmente:
Event-Time Timeline
10:00 10:05 10:10 10:15
|----------|----------|----------|
Window A
^
Watermark
Quando a watermark ultrapassa o fim da janela, o resultado pode ser finalizado e o estado correspondente removido.
Esse mecanismo é importante porque streams podem durar meses ou anos.
Sem políticas adequadas, determinadas operações stateful poderiam acumular estado indefinidamente.
15. Exemplo de agregação com Watermark
Vamos construir uma consulta que conta eventos por janelas de um minuto.
eventos = (
spark.readStream
.format("rate")
.option("rowsPerSecond", 5)
.load()
)
resultado = (
eventos
.withWatermark(
"timestamp",
"2 minutes"
)
.groupBy(
F.window(
F.col("timestamp"),
"1 minute"
)
)
.count()
)
Agora iniciamos:
query = (
resultado
.writeStream
.format("console")
.outputMode("update")
.option("truncate", False)
.trigger(processingTime="5 seconds")
.start()
)
Utilizamos update para observar resultados parciais das janelas.
Se utilizássemos append, os resultados seriam emitidos quando as janelas estivessem prontas para finalização, conforme o avanço da watermark.
Isso pode produzir uma espera perceptível.
Para encerrar:
query.stop()
16. Watermark não garante aceitar todos os eventos dentro do atraso
Existe uma sutileza importante.
O atraso configurado é utilizado para controlar a evolução da watermark e a retenção de estado.
Ele não representa uma garantia absoluta de que todo evento com atraso inferior ao intervalo será aceito em qualquer circunstância.
O comportamento depende do avanço do tempo de evento observado e da operação executada.
Além disso, o Spark pode descartar registros considerados atrasados para determinadas operações stateful.
Por isso, a escolha da watermark deve considerar:
- Distribuição real de atrasos.
- Requisitos de completude.
- Memória disponível.
- Latência desejada.
- Tolerância a correções posteriores.
Watermarks representam um compromisso entre completude, latência e consumo de estado.
17. Stateful vs. Stateless Processing
Nem toda transformação streaming precisa manter informações de eventos anteriores.
Uma transformação stateless processa registros sem depender de estado acumulado entre microbatches.
Exemplo:
filtrado = eventos.filter(
F.col("value") > 100
)
Já uma agregação precisa manter informações entre microbatches.
Exemplo:
agregado = (
eventos
.groupBy(
F.window("timestamp", "1 minute")
)
.count()
)
Conceitualmente:
STATELESS
Event → Transformation → Output
STATEFUL
Event → Transformation
|
v
State Store
|
v
Output
Operações stateful são mais complexas porque exigem gerenciamento de estado e recuperação consistente.
18. State Store
O Structured Streaming utiliza mecanismos de armazenamento de estado para determinadas operações.
O estado pode representar:
- Contagens.
- Agregações.
- Informações para deduplicação.
- Dados necessários a joins streaming.
- Outros valores mantidos entre microbatches.
Conceitualmente:
Microbatch 1
|
v
State Update
|
v
State Store
Microbatch 2
|
v
Read Previous State
|
v
Update State
|
v
State Store
O Spark oferece diferentes implementações e opções de configuração para armazenamento de estado, dependendo da versão.
O volume de estado influencia diretamente:
- Memória.
- Disco.
- Checkpoint.
- Tempo de processamento.
- Recuperação de falhas.
Por isso, operações stateful exigem atenção especial em produção.
19. Checkpoints: a base da recuperação
Imagine que uma consulta streaming esteja processando eventos continuamente.
De repente, o processo falha.
Como saber quais dados já foram processados?
Como recuperar o estado de uma agregação?
Como evitar reprocessamentos inconsistentes?
O Structured Streaming utiliza checkpoints para persistir informações necessárias à recuperação de consultas compatíveis.
Um checkpoint pode conter informações sobre:
- Progresso da consulta.
- Offsets ou posições da fonte.
- Commits.
- Estado de operações stateful.
- Metadados da execução.
A estrutura exata depende da versão e do tipo de consulta.
20. Configurando Checkpoint Location
Exemplo:
query = (
eventos_transformados
.writeStream
.format("parquet")
.option(
"path",
"/tmp/streaming/output"
)
.option(
"checkpointLocation",
"/tmp/streaming/checkpoint"
)
.outputMode("append")
.start()
)
O diretório de checkpoint deve ser persistente e acessível ao ambiente de execução.
Em produção, não devemos depender de um diretório temporário local que possa desaparecer após a reinicialização do container ou nó.
Também devemos evitar compartilhar o mesmo checkpoint entre consultas independentes.
O checkpoint representa o estado de uma consulta específica.
21. O que acontece após uma falha?
Imagine:
Microbatch 1 → Completed
Microbatch 2 → Completed
Microbatch 3 → Failure
A aplicação é reiniciada com uma configuração compatível e o mesmo checkpoint.
O Spark utiliza as informações persistidas para recuperar o progresso e o estado da consulta.
Conceitualmente:
STREAMING QUERY
|
v
CHECKPOINT
|
v
PROCESS FAILURE
|
v
RESTART
|
v
RECOVER PROGRESS
|
v
CONTINUE PROCESSING
Mas essa recuperação depende de condições importantes.
A fonte precisa permitir a recuperação das posições necessárias.
O sink precisa oferecer garantias compatíveis.
E determinadas alterações na consulta ou no schema de estado podem impedir a recuperação com o mesmo checkpoint.
Checkpoint não significa que qualquer modificação no código seja segura durante um restart.
22. Exactly-once: o que realmente significa?
Um dos termos mais utilizados em streaming é exactly-once.
Mas precisamos ser precisos.
Existem diferentes semânticas de processamento e entrega.
At-most-once
Um evento pode ser perdido, mas não é necessariamente reprocessado.
At-least-once
Um evento pode ser processado novamente após falhas.
Exactly-once
Busca garantir que o efeito lógico do processamento seja aplicado uma única vez, dentro das condições e limites da arquitetura.
Structured Streaming pode oferecer garantias end-to-end de exactly-once em determinadas combinações de fontes, operações e sinks.
Mas não podemos generalizar essa garantia para qualquer destino externo.
Por exemplo, um foreachBatch que envia mensagens para uma API pode produzir duplicidades se a aplicação falhar após a chamada externa e antes de registrar o progresso.
Nesse caso, precisamos implementar idempotência.
23. foreachBatch e idempotência
foreachBatch permite aplicar uma função a cada microbatch.
Exemplo:
def processar_batch(batch_df, batch_id):
print(f"Processando batch {batch_id}")
(
batch_df.write
.mode("append")
.parquet("/tmp/streaming/destino")
)
Iniciamos:
query = (
eventos
.writeStream
.foreachBatch(processar_batch)
.option(
"checkpointLocation",
"/tmp/streaming/checkpoint_foreach"
)
.start()
)
Atenção: esse exemplo demonstra a API, mas não implementa exactly-once no destino Parquet.
Se ocorrer uma falha entre a escrita e a confirmação do progresso, o microbatch pode ser executado novamente e gerar dados duplicados.
Uma estratégia de produção pode utilizar:
- Identificador de batch.
- Escrita transacional.
- Controle de commits.
- Chaves idempotentes.
- Deduplicação.
- Operações MERGE.
O desenho precisa considerar as garantias reais do destino.
24. Kafka e Structured Streaming
Agora vamos conectar o Spark a uma arquitetura de eventos.
Kafka é frequentemente utilizado como plataforma de mensageria e streaming.
Uma arquitetura comum:
APPLICATIONS
|
v
KAFKA
|
v
SPARK STRUCTURED STREAMING
|
v
TRANSFORMATIONS
|
v
LAKEHOUSE / ANALYTICS
O Kafka organiza mensagens em tópicos e partições.
O Spark pode consumir esses dados por meio de um conector compatível.
A integração exige dependências específicas, além de configuração de rede, segurança e permissões.
25. Exemplo de leitura Kafka
Com o conector Spark Kafka instalado:
eventos_kafka = (
spark.readStream
.format("kafka")
.option(
"kafka.bootstrap.servers",
"localhost:9092"
)
.option(
"subscribe",
"pedidos"
)
.option(
"startingOffsets",
"earliest"
)
.load()
)
O DataFrame disponibiliza campos como:
key
value
topic
partition
offset
timestamp
As colunas key e value são representadas como dados binários.
Podemos convertê-las:
mensagens = eventos_kafka.select(
F.col("key").cast("string").alias("key"),
F.col("value").cast("string").alias("payload"),
"topic",
"partition",
"offset",
"timestamp"
)
Depois podemos interpretar o payload conforme o contrato de dados.
26. Interpretando eventos JSON do Kafka
Imagine mensagens com a seguinte estrutura:
{
"pedido_id": 101,
"cliente_id": 2001,
"valor": 150.75,
"status": "PAGO"
}
Definimos o schema:
from pyspark.sql.types import (
StructType,
StructField,
LongType,
DecimalType,
StringType
)
schema_pedido = StructType([
StructField("pedido_id", LongType()),
StructField("cliente_id", LongType()),
StructField("valor", DecimalType(12, 2)),
StructField("status", StringType())
])
Aplicamos:
pedidos = (
mensagens
.withColumn(
"dados",
F.from_json(
F.col("payload"),
schema_pedido
)
)
.select(
"dados.*",
"topic",
"partition",
"offset",
"timestamp"
)
)
Agora temos um DataFrame estruturado.
Mas precisamos tratar payloads inválidos, campos ausentes e alterações de schema.
A integração Kafka–Spark não elimina as responsabilidades de Data Quality e Data Contracts estudadas na Parte 14.
27. Deduplicação em streaming
Eventos duplicados podem surgir por diferentes motivos.
Imagine:
Event ID 100
Event ID 101
Event ID 100
Se cada evento representar uma transação, processar duplicidades pode distorcer os resultados.
Structured Streaming oferece operações de deduplicação, mas a estratégia precisa considerar estado e tempo.
Uma abordagem com suporte nas versões compatíveis é:
eventos_unicos = (
eventos_com_id
.withWatermark(
"event_time",
"10 minutes"
)
.dropDuplicatesWithinWatermark(
["event_id"]
)
)
dropDuplicatesWithinWatermark está disponível em versões modernas do Spark, incluindo Spark 3.5 ou posterior.
Esse mecanismo permite limitar o estado de deduplicação utilizando a watermark.
Mas não oferece deduplicação eterna de qualquer evento.
Se o requisito exigir unicidade global por tempo indefinido, precisaremos de uma estratégia diferente.
28. Streaming Joins
Também podemos combinar dados de diferentes fontes.
Por exemplo:
Orders Stream
|
+------ JOIN ------+
| |
v v
Orders Data Customers Data
| |
+--------+---------+
|
v
Enriched Orders
Existem diferentes tipos de joins em Structured Streaming.
Um join entre stream e tabela estática possui características diferentes de um join entre dois streams.
Em joins stream-stream, frequentemente precisamos controlar o crescimento do estado utilizando restrições temporais e watermarks.
Sem limites adequados, o estado pode crescer significativamente.
Por isso, a escolha do join deve considerar:
- Semântica temporal.
- Atrasos.
- Cardinalidade.
- Volume.
- Retenção de estado.
- Requisitos de completude.
29. Backpressure e atraso de processamento
Imagine que Kafka produz eventos mais rapidamente do que Spark consegue processar.
Incoming Events: 100,000/s
Spark Capacity: 60,000/s
A diferença pode produzir acúmulo de dados pendentes.
Conceitualmente:
PRODUCER RATE
|
v
EVENT QUEUE
|
v
SPARK PROCESSING
|
v
CONSUMER OUTPUT
Se a capacidade de processamento for inferior à taxa de entrada durante muito tempo, a latência tende a aumentar.
Precisamos acompanhar:
- Taxa de entrada.
- Taxa de processamento.
- Duração dos microbatches.
- Atraso na fonte.
- Estado acumulado.
- Capacidade dos Executors.
Em determinadas fontes, podemos limitar a quantidade de dados processados por trigger.
Por exemplo, o conector Kafka oferece configurações como maxOffsetsPerTrigger.
Mas limitar a ingestão não resolve a falta estrutural de capacidade.
Apenas controla quanto trabalho é admitido por microbatch.
30. Observabilidade de Streaming
Em batch, normalmente acompanhamos a duração de cada Job.
Em streaming, precisamos monitorar uma aplicação de longa duração.
Structured Streaming disponibiliza métricas de progresso.
Exemplo:
query.lastProgress
Também podemos consultar:
query.status
E:
query.recentProgress
Entre os indicadores relevantes estão:
-
inputRowsPerSecond. -
processedRowsPerSecond. -
numInputRows. - Duração dos triggers.
- Informações de estado.
- Progresso das fontes.
- Informações do sink.
Essas métricas devem ser interpretadas em conjunto.
Uma taxa de processamento elevada não garante baixa latência se a fonte estiver acumulando backlog.
31. Exemplo de monitoramento
Podemos inspecionar o progresso de uma consulta:
progresso = query.lastProgress
if progresso is not None:
print(
"Input rows/s:",
progresso.get("inputRowsPerSecond")
)
print(
"Processed rows/s:",
progresso.get("processedRowsPerSecond")
)
print(
"Batch duration:",
progresso.get("durationMs")
)
Em produção, essas informações podem ser enviadas para uma plataforma de observabilidade.
Também devemos monitorar:
- Falhas.
- Reinicializações.
- Crescimento do checkpoint.
- Uso de memória.
- Estado.
- Latência ponta a ponta.
- Qualidade dos eventos.
A observabilidade precisa abranger tanto o Spark quanto a fonte e o destino.
32. Streaming e Lakehouse
Agora podemos conectar as Partes 13, 14 e 15.
Imagine:
OPERATIONAL DATABASE
|
v
CDC
|
v
KAFKA
|
v
SPARK STRUCTURED STREAMING
|
v
DATA CONTRACT VALIDATION
|
v
DATA QUALITY RULES
|
v
LAKEHOUSE TABLE
|
v
BI / ANALYTICS / ML
Essa arquitetura pode oferecer ingestão incremental e disponibilização contínua de dados analíticos.
Mas existem responsabilidades importantes:
- Idempotência.
- Evolução de schema.
- Controle de estado.
- Checkpoints.
- Qualidade.
- Consistência transacional.
- Monitoramento.
- Recuperação de falhas.
A combinação de Spark, Kafka e um formato Lakehouse não garante automaticamente que todos esses aspectos estejam resolvidos.
Precisamos projetá-los.
33. Streaming não é sempre a melhor escolha
Existe uma tendência de associar streaming a arquiteturas mais modernas.
Mas nem todo problema exige processamento contínuo.
Imagine um relatório financeiro atualizado uma vez por dia.
Um pipeline batch pode ser mais simples, barato e confiável.
Agora imagine uma plataforma que precisa detectar transações suspeitas em poucos segundos.
Nesse caso, streaming pode ser necessário.
A decisão deve considerar:
| Critério | Batch | Streaming |
|---|---|---|
| Latência | Normalmente maior | Potencialmente menor |
| Complexidade operacional | Geralmente menor | Geralmente maior |
| Estado contínuo | Não necessariamente | Pode ser necessário |
| Recuperação | Por execução | Por progresso e checkpoint |
| Monitoramento | Jobs e SLAs | Progresso contínuo e backlog |
| Custo | Depende da frequência | Depende da operação contínua |
Streaming não é automaticamente mais rápido ou mais barato.
Ele atende a requisitos diferentes.
34. Microbatch vs. processamento de baixa latência
O Structured Streaming oferece diferentes possibilidades de execução, mas precisamos evitar confundir os conceitos.
O microbatch é o mecanismo predominante e mais amplamente utilizado.
Existem recursos específicos para execução contínua em determinadas versões e operações, com limitações próprias.
Além disso, tecnologias especializadas em streaming podem oferecer modelos diferentes de processamento e gerenciamento de estado.
A escolha entre Spark e outras engines deve considerar:
- Latência exigida.
- Volume.
- Semântica de processamento.
- Complexidade de estado.
- Operações necessárias.
- Ecossistema.
- Custo operacional.
Se o requisito for latência de poucos milissegundos, não devemos assumir que uma arquitetura Spark baseada em microbatches seja a melhor solução.
35. Um laboratório completo de streaming local
Vamos finalizar com um exemplo independente que gera eventos, transforma os dados e grava arquivos Parquet.
Esse laboratório não exige Kafka.
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from time import sleep
spark = (
SparkSession.builder
.master("local[2]")
.appName("StreamingEndToEnd")
.getOrCreate()
)
fonte = (
spark.readStream
.format("rate")
.option("rowsPerSecond", 10)
.load()
)
eventos = (
fonte
.select(
F.col("timestamp").alias("event_time"),
F.col("value").alias("event_id"),
(F.col("value") % 5).alias("categoria_id"),
(F.col("value") * 1.5).alias("valor")
)
)
query = (
eventos
.writeStream
.format("parquet")
.outputMode("append")
.option(
"path",
"/tmp/spark-streaming-lab/output"
)
.option(
"checkpointLocation",
"/tmp/spark-streaming-lab/checkpoint"
)
.trigger(processingTime="5 seconds")
.start()
)
try:
sleep(20)
finally:
query.stop()
resultado = spark.read.parquet(
"/tmp/spark-streaming-lab/output"
)
resultado.show(10, truncate=False)
print(
"Registros gravados:",
resultado.count()
)
spark.stop()
Esse exemplo demonstra:
- Criação de uma fonte streaming.
- Transformação com DataFrames.
- Execução por microbatches.
- Escrita em Parquet.
- Configuração de checkpoint.
- Encerramento da consulta.
- Leitura batch dos resultados.
A quantidade de registros produzidos pode variar conforme o tempo de inicialização e a execução dos microbatches.
Para repetir o laboratório desde o início, utilize novos diretórios de saída e checkpoint ou limpe os diretórios de teste de maneira controlada.
Em produção, nunca elimine checkpoints de uma consulta ativa ou persistente sem avaliar as consequências para recuperação e reprocessamento.
36. Erros comuns em Structured Streaming
Alguns problemas aparecem frequentemente.
Acreditar que streaming significa tempo real instantâneo.
A latência depende do mecanismo e da infraestrutura.
Não configurar checkpoints persistentes.
Isso pode comprometer a recuperação.
Confundir Event Time com Processing Time.
Isso pode produzir agregações temporalmente incorretas.
Definir watermarks sem analisar atrasos reais.
Isso pode causar perda de eventos relevantes nas operações afetadas.
Ignorar crescimento do estado.
Operações stateful podem consumir recursos crescentes.
Assumir exactly-once para qualquer sink.
As garantias dependem da arquitetura completa.
Não implementar idempotência em foreachBatch.
Retries podem produzir efeitos duplicados.
Não monitorar backlog.
Uma consulta ativa pode estar cada vez mais atrasada.
Utilizar streaming sem necessidade de baixa latência.
Isso pode aumentar complexidade e custo desnecessariamente.
37. Como eu projetaria um pipeline streaming em produção
Antes de implementar, começaria pelas perguntas:
Qual é a latência exigida?
|
v
Qual é a fonte?
|
v
Qual é a taxa de eventos?
|
v
Existe Event Time?
|
v
Qual é a distribuição de atrasos?
|
v
Existem operações stateful?
|
v
Qual é o limite de estado?
|
v
Como funcionam checkpoints?
|
v
Qual é a garantia do sink?
|
v
Como implementar idempotência?
|
v
Como monitorar backlog?
|
v
Como recuperar falhas?
Só depois definiria a arquitetura e as configurações.
Essa abordagem reduz o risco de construir um pipeline tecnicamente funcional, mas operacionalmente frágil.
38. A principal lição deste capítulo
Structured Streaming amplia o modelo de processamento do Spark para dados que chegam continuamente.
Mas streaming confiável exige mais do que utilizar readStream e writeStream.
Precisamos compreender:
- Microbatches.
- Event Time.
- Processing Time.
- Watermarks.
- Estado.
- Checkpoints.
- Semântica de entrega.
- Idempotência.
- Observabilidade.
- Recuperação.
O princípio mais importante é:
Streaming não é apenas processar eventos rapidamente. É manter resultados corretos e recuperáveis enquanto os dados continuam chegando.
39. Como penso hoje
Quando avalio uma arquitetura streaming, não pergunto apenas:
Quantos eventos por segundo ela consegue processar?
Também quero saber:
- Qual é a latência ponta a ponta?
- Como eventos atrasados são tratados?
- Qual é o tamanho do estado?
- O que acontece após uma falha?
- Quais garantias o destino oferece?
- Como evitamos duplicidades?
- Como identificamos backlog?
- Como reprocessamos eventos?
- Como validamos contratos e qualidade?
- Qual é o custo operacional?
Essas perguntas representam a diferença entre uma demonstração de streaming e uma plataforma de dados confiável.
Próximo artigo
Agora entendemos os fundamentos do Structured Streaming.
Na próxima parte, vamos aprofundar um dos cenários mais importantes de integração de dados:
Apache Kafka, CDC e pipelines incrementais com Spark.
Vamos explorar:
- Arquitetura Kafka.
- Topics, Partitions e Offsets.
- Consumer Groups.
- CDC com Debezium.
- Integração com PostgreSQL.
- Eventos de INSERT, UPDATE e DELETE.
- Ordenação e deduplicação.
- Evolução de schema.
- Upserts em Lakehouse.
- Idempotência.
- Replay e recuperação.
- Observabilidade.
- Arquitetura end-to-end.
A pergunta central será:
Como capturar mudanças de sistemas transacionais e disponibilizá-las em uma plataforma analítica sem depender de cargas completas?
Série: Como aprendi Apache Spark
← Parte 14: Data Quality, Data Contracts e Governança
Próximo: Parte 16 — Apache Kafka, CDC e pipelines incrementais com Spark
Top comments (0)