(+351) 21 24 10006  ·  info@bconcepts.pt
Carnaxide, Lisboa
Lakehouse

Como criar tabelas Delta CDC em Lakehouse: passo a passo

João Barros 07 de September de 2026 6 min de leitura

Vamos implementar um fluxo simples de Change Data Capture (CDC) usando tabelas Delta em Lakehouse para capturar inserts, updates e deletes de uma fonte transacional. Isto é útil para manter sincronização eficiente entre sistemas e permitir análises históricas sem replicar toda a base de dados. Com um fluxo CDC bem desenhado podes reduzir o volume de dados movidos: por exemplo, em vez de reprocessar 100.000 registos por hora, só aplicas as 1.200 mudanças ocorridas nesse período.

Pré-requisitos

  • Conta com acesso ao Microsoft Fabric e um Lakehouse configurado.
  • Permissões para criar tabelas Delta e executar notebooks com PySpark.
  • Fonte de dados (CSV ou stream) com um identificador único e um campo de operação (op: I/U/D) ou um log de alterações.

Adicionalmente, é recomendável ter estimativas de volume: por exemplo, saber que esperas em média 500–5.000 eventos CDC por hora ajuda a definir micro-batches e a configuração do cluster. Garantir que tens políticas de retenção e limpeza (VACUUM) alinhadas com as necessidades de auditoria é também importante — por exemplo, retenção mínima de 7 dias para Delta.

Passo 1: Preparar a tabela Delta alvo (do tipo base)

Cria uma tabela Delta que servirá de materializado com o estado actual dos registos. Vamos usar um esquema simples: id, valor, updated_at. Esta tabela manterá o último estado por id. Para um cenário real, a tabela pode ter milhares a milhões de linhas; neste exemplo inicial podes começar com 10.000 registos.

from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType

schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("valor", StringType(), True),
    StructField("updated_at", TimestampType(), True)
])

# Path do Lakehouse onde guardar a tabela
target_path = "/lakehouse/yourcatalog/yourdb/delta_target"

# Criar DataFrame vazio e gravar como Delta
empty_df = spark.createDataFrame([], schema)
empty_df.write.format("delta").mode("overwrite").save(target_path)

# Criar tabela no catálogo (opcional)
spark.sql(f"CREATE TABLE IF NOT EXISTS yourdb.delta_target USING DELTA LOCATION '{target_path}'")

Dica: se esperas muitos registos, considera particionar por uma coluna adequada (ex.: ano_mês) ou usar Z-order para optimizar leituras por id.

Passo 2: Ingestão inicial (full load)

Carrega o estado inicial da fonte para a tabela Delta. Isto garante que a tabela alvo começa com um snapshot consistente. Numa base de produção, um full load pode demorar minutos a horas (ex.: 1 milhão de registos), por isso faz isto fora de horas ou com throttling.

# Exemplo a partir de um CSV de origem
source_df = spark.read.format("csv").option("header", True).schema(schema).load("/data/initial_snapshot.csv")

# Gravar no target (substitui conteúdo inicial)
source_df.write.format("delta").mode("overwrite").save(target_path)

Após o full load valida: por exemplo, confirma que existem 10.000 registos com spark.read.format("delta").load(target_path).count() ou executa checksums por amostra para garantir integridade.

Passo 3: Estruturar o feed CDC

Assume-se um feed com colunas: id, valor, op, event_time. op = 'I'|'U'|'D'. Prepara um DataFrame com as mudanças a aplicar. Numa janela de 1 hora podes ter, por exemplo, 300 updates, 150 inserts e 50 deletes — estes números ajudam a definir o tamanho dos batches e a memória necessária.

cdc_schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("valor", StringType(), True),
    StructField("op", StringType(), False),
    StructField("event_time", TimestampType(), True)
])

cdc_df = spark.read.format("csv").option("header", True).schema(cdc_schema).load("/data/cdc_feed.csv")

Valida a ordem e a exactidão dos timestamps. Se o feed vier desordenado, usa uma estratégia de deduplicação baseada em event_time.

Passo 4: Aplicar CDC com MERGE (upsert/delete)

Usa MERGE INTO em Delta para aplicar inserts, updates e deletes de forma transaccional. Filtra por ordem de eventos se necessário (ex.: último evento por id). O MERGE garante atomicidade — ou seja, ou todas as alterações do batch entram com sucesso, ou nenhuma entra.

from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, target_path)

# Optional: se existirem múltiplos eventos por id, pega o último por event_time
from pyspark.sql import Window
from pyspark.sql.functions import row_number

w = Window.partitionBy("id").orderBy(cdc_df["event_time"].desc())
cdc_latest = cdc_df.withColumn("rn", row_number().over(w)).filter("rn = 1").drop("rn")

# Executar MERGE
(delta_table.alias("t")
  .merge(cdc_latest.alias("s"), "t.id = s.id")
  .whenMatchedUpdate(
      condition = "s.op != 'D'",
      set = {"valor": "s.valor", "updated_at": "s.event_time"}
  )
  .whenMatchedDelete(condition = "s.op = 'D'")
  .whenNotMatchedInsert(values = {"id": "s.id", "valor": "s.valor", "updated_at": "s.event_time"})
  .execute())

Exemplo prático: num batch com 500 eventos deduplicados, esperas aplicar ~350 updates, 120 inserts e 30 deletes. Após o MERGE, valida o número de linhas e alguns registos aleatórios para confirmar o comportamento.

Passo 5: Automatizar e tratar erros comuns

Coloca este fluxo num notebook agendado ou numa pipeline. Lida com erros comuns: dados duplicados no feed, falta de campo op, conflitos de timestamp. Implementa logs e métricas (n.º aplicados, n.º rejeitados) para monitorização. Se usares Microsoft Fabric, agenda o notebook com triggers e captura falhas para reprocessamento.

# Exemplo simples de validação antes do MERGE
invalid_ops = cdc_df.filter("op NOT IN ('I','U','D')").count()
if invalid_ops > 0:
    raise ValueError(f"Encontradas {invalid_ops} operações inválidas no feed CDC")

Outras práticas: usar checkpoints para structured streaming, limitar tamanho dos batches a 100k eventos, e configurar VACUUM com retenção mínima (ex.: 7 dias) para evitar remoção prematura de históricos necessários para replays.

Verificar o resultado

Confirma que os registos estão correctos consultando a tabela Delta e comparando com o feed esperado. Verifica inserts, updates e deletes e o campo updated_at. Exemplos de verificação: contar diferenças entre snapshot esperado e tabela (por id), verificar timestamps mais recentes e confirmar que não existem linhas marcadas como eliminadas quando não deveriam.

# Consultar a tabela Delta
spark.read.format("delta").load(target_path).show()

# Comparar contagens
print("Total na tabela:", spark.read.format("delta").load(target_path).count())

Conclusão

Implementaste um fluxo básico de CDC com tabelas Delta em Lakehouse, usando MERGE para garantir atomicidade. Próximos passos: integrar um stream (structured streaming) para reduzir latência, armazenar metadados de lineage e implementar testes automáticos e alertas. Dica prática: começa por pequenos lotes de CDC (ex.: 1k–5k eventos) e valida sempre a ordem dos eventos antes de escalar para produção.

← Voltar a Lakehouse Marcar consultoria gratuita →