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: