Pare de codificar manualmente os pipelines de captura de dados alterados

Experimentei o AutoCDC de Snapshots em Python e fiquei surpreso ao ver como 4 linhas de código poderiam substituir o que eu estava fazendo em 1.500 linhas de código antes. — Engenheiro de dados sênior, Fortune 500 Aerospace & Protection Firm

Alterar captura de dados (CDC) e dimensões mudando lentamente (SCD) são fundamentais para análises modernas e cargas de trabalho de IA. As equipes contam com eles para manter as tabelas downstream precisas à medida que os dados operacionais mudam, seja isso significando manter uma visão atual do negócio ou preservar o contexto histórico completo.

No entanto, na prática, os gasodutos CDC são frequentemente alguns dos gasodutos mais difíceis de construir e operar. As equipes rotineiramente realizam testes complexos MERGE lógica para lidar com atualizações, exclusões e dados que chegam tarde: camadas em tabelas de preparação, funções de janela e suposições de sequenciamento que são difíceis de raciocinar e ainda mais difíceis de manter à medida que os pipelines evoluem. Embora os LLMs possam tornar a produção desse código mais rápida, eles não reduzem a complexidade de acertar ou mantê-lo correto ao longo do tempo – eles podem gerar código, mas não entendem seus dados.

Isso é exatamente o que CDC automático foi projetado para resolver. Em vez de codificar essa lógica diretamente em cada pipeline, as equipes declaram a semântica desejada e a plataforma a implementa. Código do Gênio pode então construir sobre essa base para gerar pipelines que sejam corretos por design.

Nesta postagem, examinaremos os padrões CDC e SCD que engenheiros de dados e profissionais de SQL encontram todos os dias, por que esses padrões são difíceis de implementar manualmente e como o AutoCDC em Lakeflow Spark Declarative Pipelines os automatiza declarativamente – ao mesmo tempo que oferece melhorias significativas em preço e desempenho.

CDC e SCD ainda são difíceis para engenheiros de dados

Mesmo para equipes que entendem bem esses padrões, acertá-los e mantê-los corretos ao longo do tempo é onde as coisas falham. À medida que os volumes de dados crescem e os casos de utilização se expandem, os pipelines tornam-se frágeis; problemas de correção surgem tarde; e mesmo pequenas alterações exigem reescritas cuidadosas para evitar corromper tabelas downstream.

Mantendo tabelas SCD Tipo 1

As tabelas SCD Tipo 1 substituem as linhas existentes para refletir o estado mais recente. Mesmo este caso “simples” rapidamente enfrenta desafios:

• As atualizações chegam fora de ordem
• Eventos duplicados devem ser desduplicados de forma consistente
• As exclusões devem ser aplicadas corretamente
• A lógica deve permanecer idempotente durante novas tentativas e reprocessamento

O que muitas vezes começa como um simples MERGE INTO evolui para uma lógica profundamente aninhada com tabelas de preparação, funções de janela e suposições de sequenciamento que são difíceis de raciocinar (ou alterar com segurança). Com o tempo, as equipes ficam relutantes em tocar nesses pipelines.

Manutenção do histórico de SCD Tipo 2

SCD Tipo 2 introduz complexidade adicional:

• Acompanhamento de versões de linha e janelas de validade
• Lidar com atualizações que chegam atrasadas sem corromper o histórico
• Garantir que exista exatamente uma versão “atual” a qualquer momento

Os erros aqui nem sempre falham em voz alta. Eles geralmente surgem semanas depois como um desvio métrico sutil ou a necessidade de reconstruir totalmente as tabelas históricas.

Extraindo dados alterados de diferentes fontes

Nem todos os sistemas emitem logs CDC limpos. Alguns sistemas emitem feeds de dados de alteração nativos, enquanto outros não – muitas vezes porque a equipe que consome os dados não controla o banco de dados upstream – forçando as equipes a reconstruir as alterações comparando instantâneos sucessivos de uma tabela de origem.

Apoiar ambos normalmente significa lógica separada de ingestão e processamento; diferentes suposições de correção; e mais caminhos de código para manter e depurar.

Operando pipelines CDC ao longo do tempo

Mesmo quando um pipeline de CDC está correto, ele ainda precisa sobreviver a reprocessamentos e preenchimentos, evolução de esquema, falhas e reinicializações. A lógica CDC elaborada manualmente tende a tornar-se mais frágil ao longo do tempo, à medida que estas realidades se acumulam, aumentando o risco operacional e o custo de manutenção.

Automatizando padrões complexos de CDC com engenharia de dados declarativos

CDC automático foi projetado para padronizar esses padrões comuns de CDC e SCD por trás de uma abstração declarativa. Em vez de codificação handbook como mudanças devem ser aplicadas, declaram as equipes que semântica eles desejam, e a plataforma gerencia pedidos, estado e processamento incremental.

Isso dá às equipes uma maneira consistente e repetível de implementar CDC e SCD em pipelines, em vez de reinventar o padrão a cada vez (o que é realmente o valor central da programação declarativa em geral, e Spark Declarative Pipelines especificamente). Essa base também permite que o Genie Code gere e gerencie esses pipelines usando o AutoCDC, garantindo que o desenvolvimento assistido por IA se baseie em uma semântica padronizada e pronta para produção.

Ao processar registros alterados de um feed de dados alterados (CDF), o AutoCDC manipula automaticamente registros fora de sequência e aplica atualizações corretamente com base em uma coluna de sequenciamento declarada. Para mostrar como isso funciona na prática, vamos considerar o exemplo de feed do CDC abaixo:

Lembre-se, você deve escolher SCD Tipo 1 para manter apenas os dados mais recentes ou escolher SCD Tipo 2 para manter dados históricos. Vamos começar com o Tipo 1.

Automatizando a manutenção do SCD Tipo 1 (alterar fontes de feed de dados)

Neste exemplo, um feed de dados alterados contém inserções, atualizações e exclusões para uma tabela de usuário. O objetivo é manter um visualização atual de cada registroonde novas atualizações substituem valores mais antigos.

Tabela de saída para SCD Tipo 1

O usuário 123 (Isabel) foi excluído, portanto não aparece na saída. O usuário 125 (Mercedes) mostra apenas a cidade mais recente (Guadalajara) porque o SCD Tipo 1 substitui os valores anteriores.

Com uma abordagem tradicional, isso requer MERGE lógica para desduplicar eventos, impor pedidos, aplicar exclusões e garantir que o pipeline permaneça correto durante novas tentativas ou dados que chegam atrasados. O AutoCDC substitui essa lógica frágil por uma definição de pipeline declarativa que lida automaticamente com sequenciamento, desduplicação, dados atrasados ​​e processamento incremental – eliminando dezenas de linhas de lógica de mesclagem personalizada.

Veja o exemplo de código completo em apêndice

Automatizando o histórico de SCD Tipo 2 (alterar fontes de feed de dados)

Em muitos sistemas analíticos, manter apenas o estado mais recente não é suficiente – as equipes precisam de um relatório completo história de como os registros mudam ao longo do tempo. Este é o padrão SCD Tipo 2, onde cada versão de um registro é armazenada com janelas de validade indicando quando estava ativo.

Tabela de saída para SCD tipo 2:

A tabela preserva o histórico completo. O usuário 123 possui duas versões (termina na sequência 6 quando excluído). O usuário 125 possui três versões que mostram as mudanças na cidade. Registros com __END_AT = NULL estão atualmente ativos.

Implementar isso manualmente requer várias etapas MERGE lógica para fechar registros anteriores, inserir novas versões e garantir que apenas uma versão permaneça ativa por vez. O AutoCDC automatiza essas transições de forma declarativa, gerenciando colunas de histórico e lógica de controle de versão automaticamente, garantindo a correção mesmo quando as atualizações chegam fora de ordem.

Veja o exemplo de código completo em apêndice

Inferindo o CDC a partir de fontes de snapshot

Nem todos os sistemas de origem emitem logs de alterações. Em muitos casos, as equipes recebem instantâneos periódicos de uma tabela de origem e deve inferir o que mudou entre as execuções.

Tradicionalmente, isso requer a comparação handbook de snapshots para detectar inserções, atualizações e exclusões antes de aplicar essas alterações com a lógica MERGE. O AutoCDC trata o CDC baseado em snapshots como um padrão de primeira classe, detectando automaticamente alterações em nível de linha entre snapshots e aplicando-as de forma incremental sem exigir lógica de comparação personalizada ou gerenciamento de estado.

A implementação handbook requer a detecção de alterações em nível de linha entre instantâneos, o fechamento de registros anteriormente ativos e a inserção de novas versões com janelas de validade atualizadas. O AutoCDC deriva automaticamente essas alterações e aplica a semântica do SCD Tipo 2, mantendo o histórico de versões sem exigir lógica de mesclagem de várias etapas ou rastreamento de estado de instantâneo personalizado.

Gerenciando pedidos, estado e reprocessamento

Pipelines declarativos Lakeflow Spark rastreia automaticamente o progresso incremental e lida com dados fora de sequência. Os pipelines podem se recuperar de falhas, reprocessar dados históricos e evoluir ao longo do tempo sem aplicar duas vezes ou perder alterações.

Praticamente, isso elimina a necessidade de as equipes gerenciarem a lógica de sequenciamento, a contabilidade da marca d’água ou a segurança do reprocessamento – a plataforma cuida disso.

O que há de novo: grandes ganhos de preço e desempenho

Além de simplificar a lógica do pipeline, as recentes melhorias do Databricks Runtime proporcionaram ganhos substanciais em desempenho e eficiência de custos para cargas de trabalho do AutoCDC – desde novembro de 2025:

Melhorias de desempenho e custos do SCD tipo 1

Melhorias de desempenho e custos do SCD tipo 2

Esses ganhos são importantes para pipelines do mundo actual que funcionam continuamente em escala. Enquanto MERGE INTO continua sendo uma primitiva elementary do Spark, o AutoCDC se baseia nele para lidar com dados fora de sequência e processamento incremental com mais eficiência à medida que os volumes de dados crescem.

Resultados reais com AutoCDC

As equipes que executam pipelines de CDC e SCD em produção relatam consistentemente um valor significativo do AutoCDC:

A Navy Federal Credit score Union usa AutoCDC em Lakeflow Spark Declarative Pipelines para potencializar o processamento de eventos em grande escala e em tempo actual – lidando com bilhões de eventos de aplicativos continuamente, eliminando o código CDC personalizado e a manutenção contínua do pipeline.

A simplicidade do modelo de programação Spark Declarative Pipelines combinada com seus recursos de serviço resultou em um tempo de resposta incrivelmente rápido. — Jian (Miracle) Zhou, Gerente Sênior de Engenharia, União de Crédito Federal da Marinha

Block usa AutoCDC em Lakeflow Spark Declarative Pipelines para simplificar a captura de dados alterados e pipelines de streaming em tempo actual no Delta Lake, substituindo CDC codificado manualmente e lógica de mesclagem por uma abordagem declarativa que é rápida de implementar e fácil de operar.

Com a adoção do Spark Declarative Pipelines, o tempo necessário para definir e desenvolver um pipeline de streaming passou de dias para horas. — Yue Zhang, engenheiro de software program da equipe, fundações de dados, Block

O Valora Group, um fornecedor líder de “veniência alimentar” com sede na Suíça, usa AutoCDC em Lakeflow Spark Declarative Pipelines para agilizar a captura de dados alterados para dados mestres e análises de varejo em tempo actual, substituindo o código CDC personalizado por uma abordagem declarativa que é fácil de implementar, repetir e escalar entre equipes.

Ganhamos muito fazendo CDC em SDP, porque você não escreve nenhum código – tudo é abstraído em segundo plano. O AutoCDC minimiza o número de linhas… é muito fácil de fazer. — Alexane Rose, arquiteta de dados e IA, Valora Holding

Comece

As APIs AutoCDC estão disponíveis como parte de Pipelines declarativos Lakeflow Spark no Databricks e também pode ser criado e gerenciado usando Databricks Código do Gênio.

Para saber mais:

Experimente o AutoCDC em seus próprios pipelines e elimine a lógica do CDC enrolada manualmente!

Apêndice

Exemplo de SCD Tipo 1

from delta.tables import DeltaTable
from pyspark.sql.features import max_by, struct

# Deduplicate: maintain newest file per userId
updates = (spark.learn.desk("cdc_data.customers")
    .groupBy("userId")
    .agg(max_by(struct("*"), "sequenceNum").alias("row"))
    .choose("row.*"))

# Apply SCD Sort 1: upsert updates, delete deletions
(DeltaTable.forName(spark, "goal")
    .alias("t")
    .merge(updates.alias("s"), "s.userId = t.userId")

    .whenMatchedDelete(situation="s.operation = 'DELETE'")
    .whenMatchedUpdate(
        situation="s.sequenceNum > t.sequenceNum",
        set={"title": "s.title", "metropolis": "s.metropolis", "sequenceNum": "s.sequenceNum"}
    )
    .whenNotMatchedInsertAll(situation="s.operation != 'DELETE'")
    .execute())

from pyspark import pipelines as dp
from pyspark.sql.features import col, expr

@dp.view
def customers():
    return spark.readStream.desk("cdc_data.customers")

dp.create_streaming_table("goal")

dp.create_auto_cdc_flow(
    goal="goal",
    supply="customers",
    keys=("userId"),
    sequence_by=col("sequenceNum"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    stored_as_scd_type=1
)

Exemplo de SCD Tipo 2

from delta.tables import DeltaTable
from pyspark.sql.features import col, lit, max_by, struct

# Deduplicate: maintain newest file per userId
updates = (spark.learn.desk("cdc_data.customers")
    .groupBy("userId")
    .agg(max_by(struct("*"), "sequenceNum").alias("row"))
    .choose("row.*"))

# Step 1: shut out lively rows for information being up to date or deleted
(DeltaTable.forName(spark, "goal")
    .alias("t")
    .merge(
        updates.alias("s"),
        "s.userId = t.userId AND t.__END_AT IS NULL AND s.sequenceNum > t.__START_AT"
    )
    
    .whenMatchedUpdate(set={"__END_AT": "s.sequenceNum"})
    .execute())

# Step 2: insert new rows for inserts and updates (not deletes)
new_rows = (updates
    .filter("operation != 'DELETE'")
    .withColumn("__START_AT", col("sequenceNum"))
    .withColumn("__END_AT", lit(None).solid("lengthy"))
    .drop("operation"))

new_rows.write.mode("append").saveAsTable("goal")
                    

dp.create_auto_cdc_flow(
    goal="goal",
    supply="customers",
    keys=("userId"),
    sequence_by=col("sequenceNum"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    stored_as_scd_type=2
)