DEV Community

Cover image for Como aprendi Apache Spark — Parte 16: Apache Kafka, CDC e pipelines incrementais
renanpyd
renanpyd

Posted on

Como aprendi Apache Spark — Parte 16: Apache Kafka, CDC e pipelines incrementais

#python #spark #bigdata #sql

Como aprendi Apache Spark — Parte 16: Apache Kafka, CDC e pipelines incrementais

Nota histórica: esta série revisita minha trajetória em Engenharia de Dados iniciada em 2012. Tecnologias como Structured Streaming, Debezium e os formatos Lakehouse modernos surgiram ou amadureceram posteriormente. Este capítulo conecta aquela experiência aos padrões atuais de integração de dados.

Nos capítulos anteriores, construímos uma visão progressiva do Apache Spark.

Começamos pela execução distribuída, passamos por partições, shuffle, memória, otimização de consultas, armazenamento colunar, Lakehouse e qualidade de dados.

Na Parte 15, entramos no universo do Structured Streaming.

Aprendemos como o Spark processa fluxos de eventos utilizando microbatches, checkpoints, watermarks e gerenciamento de estado.

Agora vamos responder a uma pergunta bastante comum em projetos corporativos:

Como manter uma plataforma analítica atualizada quando os dados estão mudando continuamente em sistemas transacionais?

Imagine um e-commerce.

Seu banco PostgreSQL recebe milhares de operações:

  • Novos pedidos.
  • Alterações de status.
  • Atualizações cadastrais.
  • Cancelamentos.
  • Exclusões.

Uma estratégia tradicional seria executar periodicamente:

SELECT *
FROM pedidos;
Enter fullscreen mode Exit fullscreen mode

Mas, conforme a tabela cresce, reler todos os registros se torna cada vez mais caro.

Além disso, essa abordagem pode não identificar corretamente exclusões ou alterações intermediárias.

Precisamos de um mecanismo que permita capturar as mudanças.

É aqui que entram Change Data Capture, Apache Kafka e Spark Structured Streaming.

1. Full Load vs. Incremental Load

Vamos começar pelo problema.

Imagine uma tabela com 100 milhões de registros.

Todos os dias, apenas 200 mil registros são inseridos ou alterados.

Uma carga completa processaria:

100,000,000 records
Enter fullscreen mode Exit fullscreen mode

Uma carga incremental poderia processar aproximadamente:

200,000 changed records
Enter fullscreen mode Exit fullscreen mode

A diferença potencial é significativa.

Mas precisamos entender como identificar as alterações.

Uma abordagem comum utiliza uma coluna como:

updated_at
Enter fullscreen mode Exit fullscreen mode

Exemplo:

SELECT *
FROM pedidos
WHERE updated_at > :ultimo_processamento;
Enter fullscreen mode Exit fullscreen mode

Esse mecanismo pode funcionar em determinados sistemas.

Entretanto, possui limitações:

  • Depende da atualização correta do timestamp.
  • Pode perder exclusões físicas.
  • Exige tratamento cuidadoso de empates e limites temporais.
  • Pode perder alterações se o controle incremental for implementado incorretamente.
  • Pode exigir consultas frequentes ao banco operacional.

Uma alternativa é capturar mudanças a partir do log transacional.

2. O que é Change Data Capture?

Change Data Capture (CDC) é um conjunto de técnicas para identificar alterações realizadas em uma fonte de dados.

Essas alterações podem incluir:

INSERT
UPDATE
DELETE
Enter fullscreen mode Exit fullscreen mode

Em vez de consultar repetidamente a tabela inteira, um mecanismo CDC pode observar as mudanças registradas pelo banco.

Conceitualmente:

              POSTGRESQL
                  |
                  v
           TRANSACTION LOG
                  |
                  v
              CDC ENGINE
                  |
                  v
             CHANGE EVENTS
Enter fullscreen mode Exit fullscreen mode

Existem diferentes implementações de CDC.

Algumas utilizam consultas periódicas.

Outras utilizam logs de transação.

Neste artigo, vamos concentrar nossa atenção em CDC baseado em logs, especialmente com PostgreSQL e Debezium.

3. PostgreSQL e Write-Ahead Log

O PostgreSQL utiliza um mecanismo chamado Write-Ahead Logging (WAL).

O WAL registra informações necessárias à recuperação e à durabilidade das transações.

Ele também pode ser utilizado, por meio de decodificação lógica, para disponibilizar alterações a sistemas externos.

Conceitualmente:

SQL TRANSACTION
      |
      v
POSTGRESQL
      |
      v
     WAL
      |
      v
LOGICAL DECODING
      |
      v
CHANGE EVENTS
Enter fullscreen mode Exit fullscreen mode

A captura lógica exige configurações e permissões apropriadas.

No PostgreSQL, isso envolve recursos como:

  • Logical Replication.
  • Replication Slots.
  • Publications.
  • Configurações de WAL.
  • Controle de retenção.

Uma preocupação operacional importante é que replication slots podem reter WAL quando o consumidor não avança.

Se esse comportamento não for monitorado, o armazenamento do banco pode crescer significativamente.

Portanto, CDC não é apenas uma configuração de integração.

É também uma responsabilidade operacional.

4. O papel do Debezium

O Debezium é uma plataforma de captura de alterações de bancos de dados.

Uma arquitetura comum utiliza Debezium com Kafka Connect.

                  POSTGRESQL
                      |
                      v
                 LOGICAL WAL
                      |
                      v
                   DEBEZIUM
                      |
                      v
                 KAFKA CONNECT
                      |
                      v
                  KAFKA TOPIC
Enter fullscreen mode Exit fullscreen mode

O Debezium interpreta alterações da origem e publica eventos estruturados.

Dependendo da configuração, também pode executar um snapshot inicial dos dados antes de acompanhar as mudanças contínuas.

Isso é importante.

Um novo pipeline CDC frequentemente precisa carregar o estado existente da tabela e depois continuar aplicando alterações.

Precisamos garantir que essa transição seja consistente.

5. O que é Apache Kafka?

O Apache Kafka é uma plataforma distribuída de streaming de eventos.

Ele permite que produtores publiquem registros e consumidores os leiam.

Conceitualmente:

PRODUCERS
    |
    v
 KAFKA
    |
    +------ CONSUMER A
    |
    +------ CONSUMER B
    |
    +------ CONSUMER C
Enter fullscreen mode Exit fullscreen mode

Kafka oferece mecanismos para:

  • Persistência de eventos.
  • Organização em tópicos.
  • Particionamento.
  • Replicação.
  • Leitura por offsets.
  • Consumo independente por diferentes aplicações.

Mas Kafka não é simplesmente uma fila tradicional.

Seu modelo permite que consumidores acompanhem posições no log e, dentro das condições de retenção, releiam eventos.

Essa característica é especialmente útil em arquiteturas de dados.

6. Topics, Partitions e Offsets

Um tópico Kafka pode ser dividido em partições.

Imagine:

TOPIC: pedidos

Partition 0
[0] [1] [2] [3] [4]

Partition 1
[0] [1] [2] [3]

Partition 2
[0] [1] [2] [3] [4] [5]
Enter fullscreen mode Exit fullscreen mode

Cada partição possui sua própria sequência de offsets.

O offset identifica uma posição dentro daquela partição.

Portanto:

partition = 0
offset = 10
Enter fullscreen mode Exit fullscreen mode

e:

partition = 1
offset = 10
Enter fullscreen mode Exit fullscreen mode

representam posições diferentes.

Uma propriedade importante é que Kafka preserva a ordem dos registros dentro de uma mesma partição, não uma ordem global entre todas as partições.

Essa distinção influencia diretamente o processamento CDC.

7. A importância da chave de particionamento

Imagine eventos de um pedido:

Pedido 101 → CRIADO
Pedido 101 → PAGO
Pedido 101 → ENVIADO
Enter fullscreen mode Exit fullscreen mode

Queremos preservar a sequência de alterações desse pedido.

Uma estratégia comum é utilizar:

pedido_id
Enter fullscreen mode Exit fullscreen mode

como chave Kafka.

Isso normalmente faz com que eventos com a mesma chave sejam encaminhados à mesma partição, considerando um particionador e uma configuração estáveis.

Mas existem ressalvas.

Mudanças no número de partições, no particionador ou na estratégia de chave podem afetar essa propriedade ao longo do tempo.

Além disso, a ordem do Kafka não substitui uma regra de versionamento dos eventos no destino.

Em pipelines CDC, precisamos considerar tanto a ordem de transporte quanto a ordem lógica das alterações na origem.

8. Consumer Groups

Kafka permite organizar consumidores em grupos.

Dentro de um consumer group, as partições são distribuídas entre os consumidores participantes.

Conceitualmente:

TOPIC: pedidos
     |
     +-- Partition 0 --> Consumer A
     |
     +-- Partition 1 --> Consumer B
     |
     +-- Partition 2 --> Consumer C
Enter fullscreen mode Exit fullscreen mode

Isso permite paralelismo.

Entretanto, o número de partições limita o paralelismo efetivo de leitura por consumidores de um mesmo grupo.

Se um tópico possui três partições, adicionar dez consumidores ao mesmo grupo não significa que todos receberão partições ativas simultaneamente.

Também precisamos distinguir esse modelo da integração Spark.

O conector Kafka do Structured Streaming gerencia a leitura e o progresso da consulta utilizando seus próprios mecanismos de offsets e checkpoints.

Não devemos assumir que o gerenciamento de offsets de uma aplicação Spark seja idêntico ao de um consumidor Kafka convencional.

9. Como é um evento CDC?

Vamos considerar um evento simplificado inspirado na estrutura de envelopes Debezium.

{
  "before": {
    "pedido_id": 101,
    "status": "CRIADO",
    "valor": 150.00
  },
  "after": {
    "pedido_id": 101,
    "status": "PAGO",
    "valor": 150.00
  },
  "op": "u",
  "ts_ms": 1793500000000
}
Enter fullscreen mode Exit fullscreen mode

O evento possui:

before

Estado anterior do registro, quando disponível.

after

Estado posterior do registro, quando aplicável.

op

Tipo de operação.

ts_ms

Timestamp associado ao evento, conforme a semântica do campo.

Em envelopes Debezium reais, também podemos encontrar informações adicionais sobre origem, transação e snapshots.

Os campos exatos dependem do conector, versão e configuração.

10. Operações CDC

Em um envelope Debezium típico, podemos encontrar códigos como:

Código Significado
c Create / Insert
u Update
d Delete
r Read de snapshot

O evento de snapshot merece atenção.

Ele representa uma leitura do estado existente durante a inicialização ou uma operação de snapshot.

Não deve ser tratado automaticamente como uma nova inserção de negócio.

A aplicação downstream precisa compreender a semântica desses eventos.

11. DELETE e Tombstone

Exclusões exigem cuidado especial.

Um evento de delete pode conter:

{
  "before": {
    "pedido_id": 101,
    "status": "PAGO"
  },
  "after": null,
  "op": "d"
}
Enter fullscreen mode Exit fullscreen mode

Dependendo da configuração Debezium/Kafka, também pode ser emitido um registro tombstone, com a chave do registro e valor Kafka nulo.

Esse tombstone está relacionado à semântica de compactação de logs do Kafka.

Ele não deve ser confundido com o evento CDC de exclusão.

Uma aplicação que simplesmente descarta mensagens de valor nulo precisa compreender o impacto dessa decisão.

E uma aplicação que ignora eventos d pode manter registros que já foram excluídos da origem.

12. Arquitetura end-to-end

Vamos reunir os componentes.

                  POSTGRESQL
                      |
                      v
                     WAL
                      |
                      v
                   DEBEZIUM
                      |
                      v
                 KAFKA CONNECT
                      |
                      v
                 KAFKA TOPIC
                      |
                      v
             SPARK STRUCTURED
                 STREAMING
                      |
                      v
               SCHEMA PARSING
                      |
                      v
               CDC NORMALIZATION
                      |
                      v
               DATA QUALITY
                      |
                      v
               INCREMENTAL MERGE
                      |
                      v
                LAKEHOUSE TABLE
                      |
                      v
                  SQL / BI / ML
Enter fullscreen mode Exit fullscreen mode

Cada camada possui responsabilidades específicas.

O Debezium captura alterações.

Kafka transporta e retém eventos conforme sua configuração.

Spark interpreta, transforma e valida.

O formato Lakehouse gerencia o estado da tabela analítica.

Essa separação facilita a evolução da arquitetura.

13. Lendo Kafka com Spark Structured Streaming

Vamos utilizar PySpark.

A integração exige o conector Kafka compatível com a versão do Spark e do Scala.

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("SparkKafkaCDC")
    .getOrCreate()
)

eventos_kafka = (
    spark.readStream
    .format("kafka")
    .option(
        "kafka.bootstrap.servers",
        "localhost:9092"
    )
    .option(
        "subscribe",
        "dbserver1.public.pedidos"
    )
    .option(
        "startingOffsets",
        "earliest"
    )
    .load()
)
Enter fullscreen mode Exit fullscreen mode

O endereço, o tópico e as configurações de segurança precisam corresponder ao ambiente real.

O parâmetro startingOffsets define a posição inicial para uma nova consulta.

Ao reiniciar uma consulta com checkpoint válido, o progresso persistido normalmente prevalece sobre essa configuração inicial.

14. Interpretando o payload

Kafka disponibiliza o valor da mensagem como binário.

Podemos convertê-lo para string:

from pyspark.sql import functions as F

mensagens = eventos_kafka.select(
    F.col("key").cast("string").alias("kafka_key"),
    F.col("value").cast("string").alias("payload"),
    "topic",
    "partition",
    "offset",
    "timestamp"
)
Enter fullscreen mode Exit fullscreen mode

Agora precisamos interpretar o JSON.

Vamos definir um schema simplificado.

from pyspark.sql.types import (
    StructType,
    StructField,
    LongType,
    StringType,
    DecimalType
)

schema_registro = StructType([
    StructField("pedido_id", LongType()),
    StructField("status", StringType()),
    StructField("valor", DecimalType(12, 2))
])

schema_cdc = StructType([
    StructField("before", schema_registro),
    StructField("after", schema_registro),
    StructField("op", StringType()),
    StructField("ts_ms", LongType())
])
Enter fullscreen mode Exit fullscreen mode

Esse schema corresponde ao envelope simplificado apresentado anteriormente.

Em uma integração real, devemos ajustar o parsing à serialização utilizada, incluindo eventuais envelopes adicionais de Kafka Connect, formatos Avro ou Protobuf e integração com Schema Registry.

15. Normalizando eventos CDC

Agora aplicamos from_json.

eventos_parseados = (
    mensagens
    .withColumn(
        "cdc",
        F.from_json(
            F.col("payload"),
            schema_cdc
        )
    )
)
Enter fullscreen mode Exit fullscreen mode

Precisamos obter a chave do registro.

Para INSERT e UPDATE, ela normalmente estará em after.

Para DELETE, pode estar em before.

eventos_normalizados = (
    eventos_parseados
    .filter(F.col("cdc").isNotNull())
    .select(
        F.coalesce(
            F.col("cdc.after.pedido_id"),
            F.col("cdc.before.pedido_id")
        ).alias("pedido_id"),
        F.col("cdc.after.status").alias("status"),
        F.col("cdc.after.valor").alias("valor"),
        F.col("cdc.op").alias("operacao"),
        F.col("cdc.ts_ms").alias("event_ts_ms"),
        "topic",
        "partition",
        "offset"
    )
)
Enter fullscreen mode Exit fullscreen mode

Agora temos uma representação mais adequada para processamento incremental.

Entretanto, essa normalização ainda não resolve a ordenação lógica dos eventos.

16. Por que timestamp não é suficiente para ordenar CDC?

Imagine duas alterações do mesmo pedido:

Event A
pedido_id = 101
status = PAGO

Event B
pedido_id = 101
status = CANCELADO
Enter fullscreen mode Exit fullscreen mode

Se utilizarmos apenas um timestamp de baixa precisão, os eventos podem ter o mesmo valor temporal.

Além disso, timestamps de processamento não representam necessariamente a ordem de commit na origem.

Para garantir a aplicação correta das alterações, devemos utilizar metadados de ordenação apropriados ao conector e ao banco.

No PostgreSQL, informações relacionadas à posição no WAL, como LSN, podem fazer parte dessa estratégia.

Mas é importante validar a semântica dos campos expostos pelo Debezium.

Não devemos assumir que um único campo de timestamp seja suficiente para reconstruir a ordem transacional em qualquer cenário.

17. O problema dos eventos fora de ordem

Imagine:

Source Order:

1. CRIADO
2. PAGO
3. CANCELADO
Enter fullscreen mode Exit fullscreen mode

Mas o destino recebe ou aplica:

1. CRIADO
2. CANCELADO
3. PAGO
Enter fullscreen mode Exit fullscreen mode

Se aplicarmos simplesmente a última mensagem recebida, o estado final ficará incorreto.

O pedido poderá aparecer como PAGO quando deveria estar CANCELADO.

Por isso, pipelines CDC precisam definir:

  • Chave de negócio.
  • Ordenação lógica.
  • Estratégia de deduplicação.
  • Tratamento de eventos antigos.
  • Política de reprocessamento.

A arquitetura precisa preservar ou reconstruir a sequência correta de alterações.

18. Deduplicação por chave e versão

Vamos imaginar um dataset normalizado com:

pedido_id
status
valor
source_version
Enter fullscreen mode Exit fullscreen mode

Aqui, source_version representa uma versão lógica de origem confiável, já normalizada e comparável dentro do domínio de cada pedido.

Não estamos assumindo que esse campo exista diretamente no envelope Debezium.

Ele precisa ser construído conforme a semântica real da origem.

Podemos selecionar a versão mais recente de cada pedido em um conjunto finito de eventos:

from pyspark.sql import Window
from pyspark.sql import functions as F

janela = (
    Window
    .partitionBy("pedido_id")
    .orderBy(F.col("source_version").desc())
)

eventos_recentes = (
    df_eventos
    .withColumn(
        "rn",
        F.row_number().over(janela)
    )
    .filter(F.col("rn") == 1)
    .drop("rn")
)
Enter fullscreen mode Exit fullscreen mode

Esse exemplo é adequado para um DataFrame batch, como o recebido dentro de foreachBatch.

Para resultados determinísticos, a ordenação precisa ser total ou possuir um critério de desempate definido.

Também precisamos tratar versões repetidas e eventos de delete.

19. foreachBatch para processamento incremental

Uma estratégia comum é consumir eventos com Structured Streaming e aplicar cada microbatch ao destino.

def processar_microbatch(batch_df, batch_id):
    print(
        f"Processando microbatch: {batch_id}"
    )

    # Validar dados
    # Normalizar operações
    # Deduplicar eventos
    # Aplicar MERGE transacional
Enter fullscreen mode Exit fullscreen mode

Iniciamos:

query = (
    eventos_normalizados
    .writeStream
    .foreachBatch(processar_microbatch)
    .option(
        "checkpointLocation",
        "/data/checkpoints/pedidos_cdc"
    )
    .start()
)
Enter fullscreen mode Exit fullscreen mode

O checkpoint deve utilizar armazenamento persistente e adequado ao ambiente.

Importante: foreachBatch oferece semântica de execução at-least-once para a função.

Se ocorrer uma falha, o mesmo microbatch pode ser executado novamente.

Portanto, precisamos tornar a aplicação no destino idempotente.

20. Upsert com MERGE INTO

Em um Lakehouse compatível, podemos aplicar atualizações incrementais com MERGE.

Exemplo conceitual:

MERGE INTO lakehouse.sales.pedidos AS destino
USING pedidos_incrementais AS origem
ON destino.pedido_id = origem.pedido_id

WHEN MATCHED
  AND origem.operacao = 'd'
THEN DELETE

WHEN MATCHED
  AND origem.operacao IN ('c', 'u', 'r')
THEN UPDATE SET
    destino.status = origem.status,
    destino.valor = origem.valor

WHEN NOT MATCHED
  AND origem.operacao IN ('c', 'u', 'r')
THEN INSERT (
    pedido_id,
    status,
    valor
)
VALUES (
    origem.pedido_id,
    origem.status,
    origem.valor
);
Enter fullscreen mode Exit fullscreen mode

Esse exemplo demonstra a lógica básica de upsert e delete.

Mas ainda existe um problema.

Se um evento antigo chegar depois de um evento novo, o MERGE poderá sobrescrever o estado correto.

Uma implementação robusta deve comparar versões de origem e persistir a versão aplicada.

Além disso, uma exclusão precisa ser protegida contra a ressurreição indevida de registros por eventos antigos.

Dependendo dos requisitos, isso pode exigir tombstones lógicos ou uma tabela de controle de versões.

21. Idempotência: não basta repetir o MERGE

Imagine que um microbatch seja aplicado com sucesso.

Antes de o Spark confirmar o progresso, ocorre uma falha.

Na recuperação, o mesmo microbatch é executado novamente.

Se a operação for idempotente, o estado final permanecerá correto.

Mas precisamos avaliar:

  • Se a fonte possui identificadores estáveis.
  • Se a versão da alteração é preservada.
  • Se operações antigas são ignoradas.
  • Se deletes são protegidos contra reintrodução.
  • Se o destino oferece transações adequadas.
  • Se existem efeitos colaterais externos.

Um MERGE baseado apenas em chave pode não ser suficiente.

A regra correta deve considerar a versão lógica da alteração.

22. Exemplo de MERGE protegido por versão

Imagine uma tabela com:

pedido_id
status
valor
source_version
Enter fullscreen mode Exit fullscreen mode

Podemos utilizar uma condição adicional:

MERGE INTO lakehouse.sales.pedidos AS destino
USING pedidos_incrementais AS origem
ON destino.pedido_id = origem.pedido_id

WHEN MATCHED
  AND origem.source_version > destino.source_version
THEN UPDATE SET
    destino.status = origem.status,
    destino.valor = origem.valor,
    destino.source_version = origem.source_version

WHEN NOT MATCHED
THEN INSERT (
    pedido_id,
    status,
    valor,
    source_version
)
VALUES (
    origem.pedido_id,
    origem.status,
    origem.valor,
    origem.source_version
);
Enter fullscreen mode Exit fullscreen mode

Essa estratégia demonstra a proteção contra atualizações antigas.

Entretanto, o exemplo pressupõe eventos de inserção ou atualização e uma versão de origem confiável.

Deletes precisam de tratamento adicional, porque remover fisicamente o registro também remove a versão que impediria sua reintrodução por um evento atrasado.

Em produção, a estratégia de exclusão deve ser projetada explicitamente.

23. Snapshot inicial e streaming contínuo

Um desafio importante é iniciar um pipeline CDC sobre uma tabela que já contém dados.

Imagine:

Existing Table:
10 million records
Enter fullscreen mode Exit fullscreen mode

Precisamos carregar esse estado inicial.

Depois, precisamos acompanhar as alterações que ocorrem enquanto o snapshot está sendo produzido.

Uma arquitetura CDC deve garantir uma transição consistente entre:

INITIAL SNAPSHOT
       |
       v
CONTINUOUS CHANGES
Enter fullscreen mode Exit fullscreen mode

Debezium oferece mecanismos específicos para esse cenário.

Mas o comportamento depende do conector, do modo de snapshot e da configuração.

Não devemos implementar uma carga inicial independente e simplesmente iniciar a leitura do WAL depois, sem estabelecer uma fronteira consistente entre as duas operações.

Isso pode causar perdas ou duplicidades.

24. Schema Evolution em CDC

Agora imagine que o banco operacional adiciona uma coluna:

ALTER TABLE pedidos
ADD COLUMN canal_venda VARCHAR(50);
Enter fullscreen mode Exit fullscreen mode

O evento CDC poderá passar a conter essa informação.

Mas o pipeline Spark possui um schema definido.

Se não houver uma estratégia de evolução, o novo campo poderá ser ignorado ou causar incompatibilidades, dependendo da implementação.

Precisamos planejar:

  • Evolução do schema na origem.
  • Serialização dos eventos.
  • Compatibilidade no Schema Registry, quando utilizado.
  • Evolução do parsing Spark.
  • Evolução da tabela de destino.
  • Contratos com consumidores.

Uma alteração no banco operacional pode afetar toda a cadeia analítica.

Esse é um exemplo claro da importância dos Data Contracts estudados na Parte 14.

25. Schema Registry

Em arquiteturas de eventos, um Schema Registry pode ajudar a gerenciar schemas e regras de compatibilidade.

Conceitualmente:

EVENT PRODUCER
      |
      v
SCHEMA VALIDATION
      |
      v
KAFKA TOPIC
      |
      v
SPARK CONSUMER
      |
      v
SCHEMA RESOLUTION
Enter fullscreen mode Exit fullscreen mode

Dependendo do ecossistema, podemos utilizar formatos como:

  • Avro.
  • Protobuf.
  • JSON Schema.

Mas o Schema Registry não garante automaticamente que todos os consumidores estejam semanticamente compatíveis.

Ele ajuda a gerenciar a estrutura e determinadas regras de evolução.

As regras de negócio continuam exigindo contratos e validações próprias.

26. Replay e recuperação

Uma vantagem importante do Kafka é permitir releitura de eventos ainda disponíveis conforme as políticas de retenção.

Imagine que uma transformação Spark possuía um bug.

Depois da correção, talvez precisemos reprocessar os eventos.

Conceitualmente:

KAFKA EVENT LOG
       |
       v
REPLAY
       |
       v
CORRECTED SPARK PIPELINE
       |
       v
REBUILT ANALYTICAL TABLE
Enter fullscreen mode Exit fullscreen mode

Mas replay exige planejamento.

Precisamos saber:

  • Quais offsets representam o intervalo desejado.
  • Se os eventos ainda estão disponíveis.
  • Se o destino deve ser reconstruído ou atualizado.
  • Como evitar duplicidades.
  • Como preservar a ordem.
  • Como validar o resultado.

Também não devemos apagar checkpoints indiscriminadamente para forçar um replay.

Isso pode causar efeitos inesperados no destino.

Uma estratégia segura pode utilizar uma nova consulta, um checkpoint separado e uma área de reconstrução controlada.

27. Retenção Kafka não é retenção analítica

Kafka pode reter eventos por tempo, tamanho ou políticas específicas.

Mas não devemos tratá-lo automaticamente como armazenamento histórico permanente.

Imagine que o tópico possui retenção de sete dias.

Se um pipeline ficar indisponível por dez dias, determinados eventos podem não estar mais disponíveis.

Isso pode comprometer a recuperação.

Portanto, precisamos alinhar:

KAFKA RETENTION
       |
       v
MAXIMUM RECOVERY WINDOW
       |
       v
PIPELINE SLA
Enter fullscreen mode Exit fullscreen mode

Para requisitos de histórico e auditoria, podemos persistir eventos brutos em uma camada de armazenamento durável.

Essa camada pode servir como base para reprocessamentos posteriores.

28. Observabilidade de CDC

Uma arquitetura CDC possui vários pontos de falha.

POSTGRESQL
    |
    v
DEBEZIUM
    |
    v
KAFKA
    |
    v
SPARK
    |
    v
LAKEHOUSE
Enter fullscreen mode Exit fullscreen mode

Precisamos monitorar cada camada.

PostgreSQL

  • Retenção de WAL.
  • Replication slots.
  • Lag de replicação.
  • Uso de armazenamento.
  • Disponibilidade.

Debezium / Kafka Connect

  • Estado do conector.
  • Erros de captura.
  • Progresso do snapshot.
  • Lag.
  • Falhas de serialização.

Kafka

  • Retenção.
  • Disponibilidade de brokers.
  • Partições.
  • Produção e consumo.
  • Lag, conforme o consumidor.

Spark

  • Duração dos microbatches.
  • Taxa de entrada.
  • Taxa de processamento.
  • Checkpoints.
  • Falhas.
  • Backlog.

Lakehouse

  • Tempo de MERGE.
  • Quantidade de arquivos.
  • Compactação.
  • Qualidade.
  • Consistência.
  • Freshness.

Uma aplicação Spark saudável não significa que toda a cadeia CDC esteja saudável.

29. Latência ponta a ponta

Em uma arquitetura CDC, a latência relevante geralmente começa na alteração da origem.

Conceitualmente:

Database Commit
      |
      v
CDC Capture
      |
      v
Kafka Publish
      |
      v
Spark Processing
      |
      v
Lakehouse Commit
      |
      v
Consumer Availability
Enter fullscreen mode Exit fullscreen mode

Podemos definir:

[

\text{End-to-End Latency}

\text{Availability Time}

\text{Source Commit Time}
]

Essa métrica precisa utilizar timestamps confiáveis e semântica bem definida.

O timestamp de processamento do Spark, isoladamente, não mede toda a latência da cadeia.

Também precisamos distinguir latência média de percentis como P95 e P99.

Uma média baixa pode esconder eventos que demoram muito mais do que o esperado.

30. Data Quality em eventos CDC

Nem todo evento recebido deve ser aplicado automaticamente.

Precisamos validar:

  • Identificador do registro.
  • Tipo de operação.
  • Estrutura do payload.
  • Versão da alteração.
  • Campos obrigatórios.
  • Regras de negócio.
  • Compatibilidade de schema.

Por exemplo:

eventos_validos = eventos_normalizados.filter(
    F.col("pedido_id").isNotNull()
    & F.col("operacao").isin(
        "c", "u", "d", "r"
    )
)
Enter fullscreen mode Exit fullscreen mode

Essa é apenas uma validação inicial.

Em produção, precisamos diferenciar:

  • Eventos válidos.
  • Eventos inválidos.
  • Tombstones.
  • Eventos de controle.
  • Alterações incompatíveis.

Também precisamos preservar os dados necessários para investigação.

Descartar silenciosamente eventos CDC pode comprometer a consistência do destino.

31. Laboratório local: simulando CDC sem Kafka

Agora vamos construir um exemplo executável em PySpark.

O objetivo é compreender a lógica de aplicação incremental sem depender de PostgreSQL, Kafka ou Debezium.

Vamos simular eventos com:

  • Identificador.
  • Operação.
  • Status.
  • Valor.
  • Versão de origem.

Primeiro, criamos os eventos.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql import Window
from pyspark.sql.types import (
    StructType,
    StructField,
    LongType,
    StringType,
    DecimalType
)
from decimal import Decimal

spark = (
    SparkSession.builder
    .master("local[*]")
    .appName("CDCLab")
    .getOrCreate()
)

schema = StructType([
    StructField("pedido_id", LongType()),
    StructField("operacao", StringType()),
    StructField("status", StringType()),
    StructField("valor", DecimalType(12, 2)),
    StructField("source_version", LongType())
])

eventos = [
    (101, "c", "CRIADO", Decimal("150.00"), 1),
    (101, "u", "PAGO", Decimal("150.00"), 2),
    (102, "c", "CRIADO", Decimal("230.00"), 1),
    (101, "u", "ENVIADO", Decimal("150.00"), 3),
    (102, "d", None, None, 2),
    (103, "c", "CRIADO", Decimal("99.90"), 1),
    (101, "u", "PAGO", Decimal("150.00"), 2)
]

df_eventos = spark.createDataFrame(
    eventos,
    schema=schema
)

df_eventos.show(truncate=False)
Enter fullscreen mode Exit fullscreen mode

Observe que incluímos propositalmente um evento antigo repetido do pedido 101.

32. Selecionando o estado mais recente

Agora queremos identificar o último evento lógico de cada pedido.

janela = (
    Window
    .partitionBy("pedido_id")
    .orderBy(
        F.col("source_version").desc()
    )
)

eventos_recentes = (
    df_eventos
    .withColumn(
        "rn",
        F.row_number().over(janela)
    )
    .filter(F.col("rn") == 1)
    .drop("rn")
)

eventos_recentes.show(
    truncate=False
)
Enter fullscreen mode Exit fullscreen mode

Como os eventos duplicados do pedido 101 possuem a mesma versão e o mesmo conteúdo, não há conflito de estado nesse exemplo.

Em produção, versões iguais com conteúdos diferentes precisam ser tratadas como inconsistência ou resolvidas por uma regra explícita.

33. Aplicando exclusões

Agora separamos os registros cujo último evento representa delete.

registros_ativos = (
    eventos_recentes
    .filter(F.col("operacao") != "d")
    .select(
        "pedido_id",
        "status",
        "valor",
        "source_version"
    )
)

registros_excluidos = (
    eventos_recentes
    .filter(F.col("operacao") == "d")
    .select(
        "pedido_id",
        "source_version"
    )
)
Enter fullscreen mode Exit fullscreen mode

Podemos visualizar:

registros_ativos.show()

registros_excluidos.show()
Enter fullscreen mode Exit fullscreen mode

O estado ativo esperado é:

pedido_id = 101
status = ENVIADO

pedido_id = 103
status = CRIADO
Enter fullscreen mode Exit fullscreen mode

O pedido 102 foi excluído logicamente da visão ativa.

Esse laboratório reconstrói o estado final a partir de todo o conjunto de eventos disponível.

Ele não representa, sozinho, uma implementação incremental transacional de produção.

34. Por que manter o histórico de alterações?

Uma tabela de estado atual responde:

Qual é a situação do pedido agora?

Mas não responde necessariamente:

Como o pedido chegou a esse estado?

Por isso, muitas arquiteturas mantêm duas representações.

Event History

Preserva as alterações recebidas.

Current State

Representa o estado mais recente de cada entidade.

Conceitualmente:

                  CDC EVENTS
                      |
             +--------+--------+
             |                 |
             v                 v
       EVENT HISTORY       CURRENT STATE
             |                 |
             v                 v
          AUDIT           ANALYTICS
Enter fullscreen mode Exit fullscreen mode

Essa separação pode facilitar:

  • Auditoria.
  • Replay.
  • Investigação.
  • Reconstrução de tabelas.
  • Análises temporais.

Mas o histórico precisa de políticas de retenção, privacidade e acesso.

35. Uma arquitetura de referência para produção

Vamos reunir as principais decisões.

                  OPERATIONAL SYSTEM
                         |
                         v
                     POSTGRESQL
                         |
                         v
                    LOGICAL WAL
                         |
                         v
                      DEBEZIUM
                         |
                         v
                    KAFKA TOPICS
                         |
                         v
                 SPARK STREAMING
                         |
                         v
                 CONTRACT VALIDATION
                         |
                         v
                   CDC NORMALIZATION
                         |
                         v
                 ORDERING / DEDUP
                         |
                         v
                    QUALITY GATES
                         |
              +----------+----------+
              |                     |
              v                     v
          VALID EVENTS         QUARANTINE
              |
              v
         RAW EVENT HISTORY
              |
              v
         INCREMENTAL MERGE
              |
              v
         LAKEHOUSE TABLE
              |
              v
          SQL / BI / ML
Enter fullscreen mode Exit fullscreen mode

Em uma implementação real, algumas dessas responsabilidades podem ser reorganizadas.

Por exemplo, podemos persistir o histórico bruto antes das validações de negócio, preservando eventos para auditoria e reprocessamento.

O desenho depende dos requisitos de segurança, recuperação e consistência.

36. Quando CDC não é necessário?

CDC é poderoso, mas introduz complexidade.

Talvez não seja necessário quando:

  • A tabela é pequena.
  • A atualização é pouco frequente.
  • O SLA permite cargas diárias.
  • Não existem deletes relevantes.
  • O sistema de origem oferece APIs incrementais confiáveis.
  • O custo operacional do CDC supera seus benefícios.

Em alguns cenários, uma carga incremental por timestamp bem implementada pode ser suficiente.

A decisão deve considerar:

  • Volume.
  • Frequência de mudanças.
  • Latência.
  • Custo.
  • Complexidade.
  • Garantias de consistência.
  • Capacidade operacional da equipe.

A arquitetura mais sofisticada nem sempre é a melhor arquitetura.

37. Erros comuns em pipelines CDC

Alguns erros merecem atenção especial.

Utilizar timestamp como única garantia de ordenação.

Pode não representar a sequência real das alterações.

Ignorar eventos de delete.

O destino pode manter registros inexistentes na origem.

Confundir tombstone Kafka com delete CDC.

São conceitos relacionados, mas diferentes.

Aplicar MERGE sem controle de versão.

Eventos antigos podem sobrescrever estados novos.

Não monitorar replication slots.

O WAL pode crescer e comprometer o PostgreSQL.

Não planejar snapshot inicial.

Podem ocorrer lacunas ou duplicidades entre carga inicial e streaming.

Não definir política de replay.

Falhas podem exigir reconstruções difíceis.

Assumir exactly-once em qualquer destino.

As garantias dependem da arquitetura completa.

Não versionar schemas.

Mudanças na origem podem quebrar consumidores.

Não preservar histórico quando necessário.

Isso dificulta auditoria e reprocessamento.

38. Como eu avaliaria uma arquitetura CDC

Antes de colocar um pipeline em produção, faria perguntas como:

Qual é o banco de origem?
          |
          v
Como as alterações são capturadas?
          |
          v
Como funciona o snapshot inicial?
          |
          v
Qual é a chave do registro?
          |
          v
Como preservamos a ordem?
          |
          v
Como tratamos duplicidades?
          |
          v
Como tratamos DELETE?
          |
          v
Como controlamos versões?
          |
          v
Como garantimos idempotência?
          |
          v
Como fazemos replay?
          |
          v
Qual é a retenção necessária?
          |
          v
Como monitoramos lag?
          |
          v
Como validamos a consistência final?
Enter fullscreen mode Exit fullscreen mode

Essas perguntas precisam ser respondidas antes de considerar a arquitetura pronta.

39. A principal lição deste capítulo

Apache Kafka, Debezium e Spark Structured Streaming permitem construir pipelines incrementais capazes de acompanhar mudanças em sistemas transacionais.

Mas a complexidade principal não está apenas em transportar eventos.

Ela está em preservar corretamente a semântica das alterações.

Precisamos compreender:

  • Origem das mudanças.
  • Ordenação.
  • Chaves.
  • Duplicidades.
  • Deletes.
  • Snapshots.
  • Versões.
  • Checkpoints.
  • Idempotência.
  • Recuperação.
  • Qualidade.

O princípio mais importante é:

CDC não é apenas capturar mudanças. É reconstruir e manter corretamente o estado dos dados a partir dessas mudanças.

40. Como penso hoje

Quando avalio uma plataforma que promete dados em tempo quase real, não pergunto apenas:

Qual é a latência do Kafka?

Quero saber:

  • Como as mudanças são capturadas?
  • Qual é a consistência do snapshot inicial?
  • Como eventos fora de ordem são tratados?
  • Como o destino evita atualizações antigas?
  • Como deletes são preservados?
  • O que acontece após uma falha?
  • Como os eventos podem ser reprocessados?
  • Como monitoramos a cadeia inteira?
  • Como validamos a consistência entre origem e destino?

Essas perguntas representam a diferença entre um pipeline que transporta mensagens e uma arquitetura incremental confiável.

Próximo artigo

Agora entendemos como capturar e processar mudanças de sistemas transacionais.

Na próxima parte, vamos aprofundar a construção de pipelines Spark para ambientes corporativos.

Orquestração, Airflow, CI/CD e DataOps com Apache Spark.

Vamos estudar:

  • Orquestração vs. processamento.
  • Apache Airflow.
  • DAGs.
  • Dependências.
  • Retries.
  • Backfills.
  • Idempotência.
  • Testes automatizados.
  • CI/CD.
  • Versionamento.
  • Deploy.
  • Observabilidade.
  • Estratégias de produção.

A pergunta central será:

Como transformar Jobs Spark isolados em pipelines automatizados, testáveis, observáveis e confiáveis?

Série: Como aprendi Apache Spark

← Parte 15: Structured Streaming, Watermarks e Checkpoints

Próximo: Parte 17 — Orquestração, Airflow, CI/CD e DataOps com Apache Spark

Top comments (0)