As APIs AUTO CDC: Simplifique a captura de dados de alterações com pipelines

Lakeflow Spark Declarative Pipelines (SDP) simplifica a captura de dados de alteração (CDC) com as APIs AUTO CDC e AUTO CDC FROM SNAPSHOT. Essas APIs automatizam o processamento complexo das dimensões de alteração lenta (SCD) Tipo 1 e Tipo 2, a partir de dados gerados por CDC ou instantâneos de banco de dados. A AUTO CDC API também dá suporte ao acompanhamento bitemporal, que registra alterações em duas dimensões de tempo (Beta). Para saber mais sobre o SCD Tipo 1 e o Tipo 2, consulte Alterar captura de dados e instantâneos. Para obter mais informações sobre o acompanhamento bitemporal, consulte Como o AUTO CDC bitemporal funciona.

A API que você usa depende da origem dos dados de alteração:

• AUTO CDC: Use isso quando o banco de dados de origem estiver com o feed CDC habilitado. AUTO CDC processa alterações de um fluxo de dados de alteração (CDF). Ele tem suporte nas interfaces SQL e Python do pipeline.
• AUTO CDC FROM SNAPSHOT: use quando o CDC não estiver habilitado no banco de dados de origem e somente os instantâneos estiverem disponíveis. Essa API compara instantâneos para determinar as alterações e processá-las. Ele tem suporte apenas na interface do Python.

Ambas as APIs dão suporte à atualização de tabelas usando o SCD Tipo 1 e o Tipo 2:

• Use o SCD Tipo 1 para atualizar registros diretamente. O histórico não é retido para registros atualizados.
• Use o SCD Tipo 2 para manter um histórico de registros, em todas as atualizações ou em atualizações para um conjunto especificado de colunas.

Somente para AUTO CDC, você também pode usar o armazenamento bitemporal, que estende o histórico do SCD Tipo 2 para rastrear alterações em duas dimensões de tempo: tempo de negócio e tempo do sistema. O Bitemporal está em Beta. Veja como o AUTO CDC bitemporal funciona.

As AUTO CDC APIs não têm suporte dos Pipelines Declarativos do Apache Spark.

Para obter sintaxe e outras referências, consulte AUTO CDC INTO (pipelines), create_auto_cdc_flow e create_auto_cdc_from_snapshot_flow.

Requirements

Para usar as APIs de CDC, o pipeline deve ser configurado para usar o SDP sem servidor ou as edições SDP Pro ou Advanced.

Como funciona o AUTO CDC

Para executar o processamento CDC com AUTO CDC, crie uma tabela de streaming e use a instrução AUTO CDC ... INTO no SQL ou a função create_auto_cdc_flow() no Python para especificar a origem, as chaves e a ordenação para o feed de alterações. Para obter uma explicação de como o sequenciamento e a lógica SCD funcionam, consulte Captura de dados de alteração e instantâneos. Veja os exemplos de AUTO CDC.

Para hidratação inicial de uma fonte com um feed de alterações, use AUTO CDC com um fluxo once e continue processando o feed de alterações. Consulte Replicar uma tabela RDBMS externa usando AUTO CDC.

Para obter detalhes de sintaxe, consulte AUTO CDC INTO (pipelines) ou create_auto_cdc_flow.

Como funciona o AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT determina as alterações nos dados de origem comparando instantâneos em ordem. Ele só tem suporte na interface de pipeline do Python. Você pode ler instantâneos de uma tabela Delta, arquivos de armazenamento em nuvem ou JDBC diretamente.

Para executar o processamento CDC com AUTO CDC FROM SNAPSHOT, crie uma tabela de streaming e, em seguida, use a função create_auto_cdc_from_snapshot_flow() para especificar o snapshot, as chaves e outros argumentos. Para obter detalhes sobre os dois padrões de ingestão e quando usar cada um, consulte Padrões de processamento de instantâneo. Veja os exemplos de AUTO CDC FROM SNAPSHOT.

Para obter detalhes de sintaxe, consulte create_auto_cdc_from_snapshot_flow.

Como funciona o AUTO CDC bitemporal

SCD Type 1 e Type 2 são unitemporais: eles acompanham as alterações ao longo de uma única dimensão temporal. O Bitemporal estende o histórico do SCD Tipo 2 para controlar as alterações em duas dimensões temporais e distinguir entre duas perspectivas:

• Horário comercial: quando o evento realmente aconteceu.
• Hora do sistema: quando o sistema registrou ou ingeriu o evento.

Assim como o SCD Tipo 2, o bitemporal preserva um histórico completo de registros. Ele adiciona uma segunda linha do tempo para que você possa reconstruir o que os dados mostraram e o que o sistema acreditava em qualquer momento no passado.

Por exemplo, um fundo de hedge ingere dados de ações de um sistema de origem. O preço das ações da Acme Corp muda em 1º de janeiro, mas o fundo não ingere essa atualização até 5 de janeiro. Bitemporal AUTO CDC permite que o fundo responda a duas perguntas distintas: qual era o preço real das ações da Acme Corp em 1º de janeiro (horário comercial) e qual preço o sistema acreditava quando o fundo tomou decisões de negociação em 3 de janeiro (hora do sistema). A capacidade de distinguir entre essas linhas do tempo é útil para auditoria, relatórios regulatórios e tomada de decisões financeiras.

Para habilitar o processamento bitemporal, definir STORED AS BITEMPORAL (SQL) ou stored_as_scd_type="bitemporal" (Python), use SEQUENCE BY para a coluna de tempo de negócios e use SYSTEM SEQUENCE BY para a coluna de tempo do sistema. A tabela de destino adiciona __SYSTEM_START_AT e __SYSTEM_END_AT colunas ao lado do SCD Tipo 2 __START_AT e __END_AT colunas. Para obter detalhes de sintaxe, consulte AUTO CDC INTO (pipelines) ou create_auto_cdc_flow.

Para obter uma explicação passo a passo de como inserções, atualizações, atualizações fora de ordem e exclusões afetam uma tabela bitemporal, consulte os exemplos de AUTO CDC bitemporal.

Usar várias colunas para sequenciamento

Para sequenciar por várias colunas (por exemplo, um carimbo de data/hora e um ID para quebrar vínculos), use uma STRUCT para combiná-las. A API ordena primeiro pelo primeiro campo e, em caso de empate, considera o segundo campo e assim por diante.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

sequence_by = struct("timestamp_col", "id_col")

Exemplos de AUTO CDC

Os exemplos a seguir demonstram o processamento de SCD Tipo 1 e Tipo 2 usando uma fonte de feed de dados de alteração. Os dados de exemplo criam novos registros de usuário, excluem um registro de usuário e atualizam os registros do usuário. No exemplo SCD Tipo 1, as últimas UPDATE operações chegam atrasadas e são excluídas da tabela de destino, demonstrando como eventos fora de ordem são tratados.

Veja a seguir os registros de entrada usados nestes exemplos. Esses dados são criados executando a consulta na seção Criar dados de exemplo .

Se você remover o comentário da linha final na consulta de geração de dados de exemplo, ela inserirá o seguinte registro que especifica truncar (limpar) a tabela em sequenceNum=3:

Criar dados de exemplo

Execute as instruções a seguir para criar um conjunto de dados de exemplo. Esse código não se destina a ser executado como parte de uma definição de pipeline. Execute-o na pasta de exploração do pipeline, em vez da pasta de transformações.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Processar atualizações do TIPO 1 do SCD

O SCD Type 1 mantém apenas a versão mais recente de cada registro. O exemplo a seguir lê a partir do feed de dados de alteração criado acima e aplica alterações a uma tabela de streaming de destino. O que são pipelines? para executar esse código.

Python

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

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL

CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

Depois de executar o exemplo scd tipo 1, a tabela de destino contém os seguintes registros:

O usuário 123 (Isabel) foi excluído e não aparece. O usuário 125 (Mercedes) mostra apenas a cidade mais recente (Guadalajara) porque o SCD Tipo 1 substitui os valores anteriores. O anterior UPDATE em sequenceNum=5 foi removido porque uma atualização posterior chegou em sequenceNum=6.

Depois de executar o exemplo com o registro TRUNCATE sem comentário, a tabela é apagada em sequenceNum=3. Isso significa que os registros 124 e 126 não estão na tabela e a tabela de destino final contém apenas o seguinte registro:

Processar atualizações do SCD Tipo 2

O SCD Tipo 2 preserva um histórico completo de alterações criando novas linhas para cada versão de um registro, com __START_AT e __END_AT colunas indicando quando cada versão estava ativa.

Python

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

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

Depois de executar o exemplo scd tipo 2, a tabela de destino contém os seguintes registros:

A tabela preserva o histórico completo. O usuário 123 possui duas versões (encerradas na sequência 6 quando foram excluídas). O usuário 125 tem três versões mostrando alterações na cidade. Os registros com __END_AT = null estão ativos no momento.

Rastrear um subconjunto de colunas com SCD Tipo 2

Por padrão, o SCD Type 2 cria uma nova versão sempre que qualquer valor de coluna é alterado. Você pode especificar um subconjunto de colunas a serem controladas, de modo que as alterações em outras colunas atualizem a versão atual em vigor em vez de gerar um novo registro de histórico.

O exemplo a seguir exclui a city coluna do acompanhamento de histórico:

Python

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

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

Como as alterações city não são controladas, as atualizações da cidade substituem a linha atual em vez de criar uma nova versão. A tabela de destino contém os seguintes registros:

Exemplos de AUTO CDC DE SNAPSHOT

As seções a seguir fornecem exemplos de como usar AUTO CDC FROM SNAPSHOT para processar instantâneos em tabelas de destino de SCD Tipo 1 ou Tipo 2. Para obter informações sobre quando usar essa API, consulte Captura de dados de alteração e instantâneos.

Exemplo: processar instantâneos usando o tempo de ingestão do pipeline

Use esta abordagem quando os instantâneos chegarem regularmente e em ordem, e você puder depender do carimbo de data/hora de execução de pipeline para controle de versão. Um novo instantâneo é ingerido com cada atualização do pipeline.

Você pode ler instantâneos de diversos tipos de fontes, incluindo tabelas Delta, arquivos em armazenamentos na nuvem e conexões JDBC.

Etapa 1: Criar dados de exemplo

Crie uma tabela contendo dados de captura. Execute o seguinte código de um notebook ou do Databricks SQL na pasta explorations do seu pipeline:

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.snapshot (
  userId INT,
  city STRING
);

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

Etapa 2: executar AUTO CDC FROM SNAPSHOT

O que são pipelines? para executar o código nesta etapa.

Escolha um tipo de origem para a visualização de instantâneo (o código de criação de amostra gera uma tabela Delta):

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

Todas as opções, gravar no destino

Em seguida, adicione a tabela de destino e o fluxo:

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  keys = ["userId"],
  stored_as_scd_type = 2
)

Após a execução inicial do pipeline, todos os registros são inseridos como linhas ativas:

Etapa 3: Simular um novo instantâneo e executar a simulação novamente

Atualize a tabela de origem para simular um novo instantâneo que está chegando (execute este código de um notebook ou arquivo SQL na pasta explorations do seu pipeline):

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

Executar o pipeline novamente. AUTO CDC FROM SNAPSHOT compara o novo instantâneo com o anterior e detecta que o usuário 1 foi excluído, os usuários 2 e 3 foram atualizados e os usuários 4 e 6 foram inseridos. Isso gera um feed de alterações e usa AUTO CDC para criar a tabela de saída.

Após a segunda execução com o SCD Tipo 2, a tabela de destino contém os seguintes registros:

O usuário 1 foi encerrado (excluído). Os usuários 2 e 3 têm duas versões mostrando mudanças em suas cidades. Os usuários 4 e 6 foram inseridos recentemente.

Após a segunda execução com o SCD Tipo 1, a tabela de destino mostra apenas o estado atual:

Exemplo: processar instantâneos usando funções de versão

Use essa abordagem quando precisar de controle explícito sobre a ordenação de instantâneos. Por exemplo, use essa abordagem quando vários instantâneos chegarem ao mesmo tempo ou instantâneos chegarem fora de ordem. Você grava uma função que especifica qual instantâneo processar em seguida e seu número de versão. A API processa instantâneos na ordem crescente de versão.

• Se vários instantâneos estiverem no armazenamento, todos eles serão processados em ordem.
• Se um instantâneo chegar fora de ordem (por exemplo, snapshot_3 chega depois de snapshot_4), ele será ignorado.
• Se não houver novos instantâneos, a função retornará None e não haverá nenhum processamento.

Etapa 1: Preparar arquivos de instantâneo

Crie arquivos CSV contendo dados de instantâneo e adicione-os a um volume ou local de armazenamento em nuvem. Nomeie os arquivos cronologicamente (por exemplo, snapshot_1.csv, ). snapshot_2.csv

Cada arquivo deve conter colunas para userId e city. Por exemplo:

snapshot_1.csv:

snapshot_2.csv:

Etapa 2: executar AUTO CDC FROM SNAPSHOT com uma função de versão

Crie um novo notebook e cole o seguinte código de pipeline. Então, o que são pipelines?.

from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
  snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data

  files = dbutils.fs.ls(snapshot_dir)
  snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]

  snapshot_versions = []
  for filename in snapshot_files:
    try:
      version = int(filename.replace("snapshot_", "").replace(".csv", ""))
      snapshot_versions.append(version)
    except ValueError:
      continue

  snapshot_versions.sort()

  if latest_snapshot_version is None:
    if snapshot_versions:
      next_version = snapshot_versions[0]
    else:
      return None
  else:
    next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
    if next_versions:
      next_version = next_versions[0]
    else:
      return None

  snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
  df = spark.read.format("csv").option("header", True).load(snapshot_path)
  return (df, next_version)


dp.create_streaming_table("main.cdc_tutorial.target_versioned")

dp.create_auto_cdc_from_snapshot_flow(
  target = "main.cdc_tutorial.target_versioned",
  source = next_snapshot_and_version,
  keys = ["userId"],
  stored_as_scd_type = 2
)

Após o processamento snapshot_1.csv, a tabela de destino contém os seguintes registros:

Após o processamento snapshot_2.csv, a tabela de destino contém os seguintes registros:

Etapa 3: Adicionar novos instantâneos

Adicione um novo arquivo CSV ao local de armazenamento com dados modificados (por exemplo, valores de cidade alterados, novas linhas ou linhas removidas). Em seguida, execute o pipeline novamente para processar o novo instantâneo.

Exemplos de AUTO CDC bitemporal

O exemplo a seguir cria uma tabela de destino bitemporal a partir de um pequeno conjunto de eventos de CDC sintéticos. A coluna bt contém o horário comercial, e a coluna st contém a hora do sistema.

Python

from pyspark import pipelines as dp

# Source: synthetic CDC events
dp.create_streaming_table(name="cdc_source")

@dp.append_flow(target="cdc_source", once=True)
def load_cdc_source():
  return spark.createDataFrame(
    [
      (1, "x10", "y10", 10, 100),
      (1, "x20", "y20", 20, 200)
    ],
    schema="id INT, x STRING, y STRING, bt INT, st INT",
  )

# Target: bitemporal table
dp.create_streaming_table(name="target_bitemporal")

dp.create_auto_cdc_flow(
  target = "target_bitemporal",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "bt",
  system_sequence_by = "st",
  stored_as_scd_type = "bitemporal"
)

SQL

-- Source: synthetic CDC events
CREATE OR REFRESH STREAMING TABLE cdc_source_sql;

CREATE FLOW cdc_source_sql AS INSERT INTO ONCE
  cdc_source_sql BY NAME
SELECT * FROM VALUES
  (1, 'x10', 'y10', 10, 100),
  (1, 'x20', 'y20', 20, 200)
  AS t(id, x, y, bt, st);

-- Target: bitemporal table
CREATE OR REFRESH STREAMING TABLE target_bitemporal_sql;

CREATE FLOW target_bitemporal_sql AS AUTO CDC INTO
  target_bitemporal_sql
FROM
  stream(cdc_source_sql)
KEYS
  (id)
SEQUENCE BY
  bt
SYSTEM SEQUENCE BY
  st
STORED AS
  BITEMPORAL;

As etapas a seguir explicam como uma tabela bitemporal registra inserções, atualizações, atualizações fora de ordem e exclusões para uma única empresa. A coluna de sequenciamento gera as colunas __START_AT e __END_AT (tempo de negócio), e a coluna de sequenciamento do sistema gera as colunas __SYSTEM_START_AT e __SYSTEM_END_AT (tempo do sistema):

O sistema lida com eventos que chegam em qualquer ordem em ambas as linhas do tempo. Quando um evento chega com um horário comercial ou tempo de sistema anterior do que os eventos já processados, o sistema corrige o histórico afetado em vez de acrescentar apenas ao final.

Etapa 1: Inserir

A empresa A é adicionada em 18/07/2025 10:01:00 (horário comercial), mas não é ingerida até 10:05:00 (hora do sistema).

Entrada:

Saída:

XFv1 é válido a partir das 10:01:00 sem término conhecido. O sistema tomou conhecimento desse fato no horário do sistema 10:05:00, sem horário de término conhecido.

Etapa 2: Atualizar

A empresa A foi atualizada em 18/07/2025 12:15:43 (hora de negócio), e o sistema consome o evento às 12:20:00 (horário do sistema). O sistema preserva o que acreditava antes da atualização ser conhecida e o histórico de negócios corrigido após a ingestão da atualização.

Entrada:

Saída:

Acreditava-se que xFv1 era válido das 10:01:00 sem fim conhecido, e o sistema manteve essa crença das 10:05:00 até 12:20:00. Agora se sabe que XFv1 é válido apenas até 12:15:43; um histórico corrigido vigora a partir do horário do sistema 12:20:00, sem término conhecido. XFv2 é válido a partir de 12:15:43, sem horário de término conhecido, e foi registrado no horário do sistema 12:20:00.

Etapa 3: atualização fora de ordem

Uma atualização fora de ordem chega indicando que a Empresa A foi atualizada em 18/07/2025 12:05:00 (hora de negócio), mas não é ingerida até 12:25:00 (hora do sistema). Quando uma atualização chega mais tarde no tempo do sistema, mas com um tempo de negócio anterior, o sistema corrige o histórico do tempo de negócio e preserva tanto o que registrava antes da atualização fora de ordem quanto o histórico corrigido.

Entrada:

Saída:

Acreditava-se que xFv1 era válido das 10:01:00 às 12:15:43, e essa crença agora é válida no tempo do sistema até 12:25:00. A nova atualização corrige a validade de negócio do XFv1 para terminar às 12:05:00, com um histórico corrigido que entra em vigor a partir do tempo de sistema 12:25:00. O XFv3 agora é conhecido por ser válido das 12:05:00 até 12:15:43, uma crença válida no tempo do sistema a partir das 12:25:00 sem fim conhecido.

Etapa 4: Excluir

A empresa A é excluída em 18/07/2025 12:30:00, e o sistema consome o evento às 12:30:00. Como uma operação de exclusão representa o fim da existência de negócios da entidade, o sistema não cria nenhuma linha de substituição. O XFv2 é exibido em duas linhas, preservando uma trilha de auditoria completa tanto do momento em que a empresa deixou de existir quanto do momento em que o sistema tomou conhecimento da exclusão.

Entrada:

Saída:

XFv2 era válido das 12:15:43 sem fim conhecido, e o sistema manteve essa crença das 12:20:00 às 12:30:00. Depois que a exclusão é processada, sabe-se que o XFv2 é válido apenas até 12:30:00, com um histórico corrigido vigente a partir da hora do sistema 12:30:00.

Limitações

• A coluna de sequenciamento deve ser um tipo de dados classificável. NULL Não há suporte para valores de sequenciamento.
• AUTO CDC FROM SNAPSHOT tem suporte apenas na interface de pipeline do Python; não há suporte para a interface SQL.
• Para transmitir dados do destino de um processo AUTO CDC, leia no feed de alterações. Para obter detalhes, consulte Leia um feed de dados de alteração em uma tabela de destino AUTO CDC.

Recursos adicionais

• Captura de dados de alteração e instantâneos: saiba mais sobre conceitos de CDC, instantâneos e tipos de SCD.
• Replique uma tabela RDBMS externa usando AUTO CDC: saiba como executar a hidratação inicial com um once fluxo e, em seguida, continuar processando as alterações.
• Tópicos avançados do AUTO CDC: saiba mais sobre operações de alteração em destinos AUTO CDC, leitura de feeds de dados de alteração e métricas de processamento.
• Tutorial: Criar um pipeline de ETL usando a captura de dados de alteração