As APIs AUTO CDC: captura simplificada de dados de mudanças 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. Estas APIs automatizam a complexidade de computar dimensões que mudam lentamente (SCD) Tipo 1 e Tipo 2, seja a partir de um feed CDC ou de instantâneos de bases de dados. A AUTO CDC API também suporta rastreamento bitemporal, que regista alterações ao longo de duas dimensões temporais (Beta). Para saber mais sobre SCD de Tipo 1 e de Tipo 2, consulte Captura de dados alterados e capturas instantâneas. Para saber mais sobre rastreamento bitemporal, veja Como funciona o BITEMPORAL AUTO CDC.

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

• AUTO CDC: Use isto quando a base de dados de origem tiver um feed CDC ativado. AUTO CDC processa alterações a partir de um feed de dados de alteração (CDF). É suportado tanto nas interfaces de SQL e Python do sistema de pipeline.
• AUTO CDC FROM SNAPSHOT: Use isto quando o CDC não estiver ativado na base de dados de origem e apenas os snapshots estiverem disponíveis. Esta API compara instantâneos para determinar alterações e depois processa-as. É suportado apenas na interface Python.

Ambas as APIs suportam a atualização de tabelas usando SCD Tipo 1 e Tipo 2:

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

Apenas para AUTO CDC, também é possível utilizar armazenamento bitemporal, que estende o histórico do SCD Tipo 2 para rastrear alterações em duas dimensões temporais: tempo de negócio e tempo de sistema. Bitemporal está em fase Beta. Veja como funciona o Bitemporal AUTO CDC.

As AUTO CDC APIs não são suportadas pelo Apache Spark Declarative Pipelines.

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

Requerimentos

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

Como funciona o AUTO CDC

Para realizar o processamento CDC com AUTO CDC, crie uma tabela de streaming e depois use a instrução AUTO CDC ... INTO em SQL ou a função create_auto_cdc_flow() em Python para especificar a fonte, as chaves e o sequenciamento para o feed de alterações. Para uma explicação de como funcionam a sequenciação e a lógica SCD, veja Alterar captura de dados e instantâneos. Veja os exemplos do AUTO CDC.

Para a inicialização inicial de uma fonte com um feed de alterações, use AUTO CDC com um once fluxo e depois continue a processar o feed de alterações. Veja Replicar uma tabela RDBMS externa usando AUTO CDC.

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

Como funciona o AUTO CDC do SNAPSHOT

AUTO CDC FROM SNAPSHOT determina alterações nos dados de origem comparando instantâneos por ordem. É suportado apenas na interface de pipeline Python. Você pode ler capturas instantâneas de uma tabela Delta, arquivos de armazenamento na cloud ou JDBC diretamente.

Para realizar o processamento CDC com AUTO CDC FROM SNAPSHOT, crie uma tabela de fluxo contínuo e, depois, 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âneos. Veja os exemplos do AUTO CDC FROM SNAPSHOT.

Para detalhes da sintaxe, consulte create_auto_cdc_from_snapshot_flow.

Como funciona o Bitemporal AUTO CDC

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

• Horário de negócios: quando o evento realmente aconteceu.
• Tempo do sistema: quando o sistema registou ou ingeriu o evento.

Tal como o SCD Tipo 2, o bitemporal preserva uma história completa dos registos. Adiciona uma segunda linha temporal para que possas reconstruir tanto o que os dados mostraram como o que o sistema acreditava em qualquer momento do passado.

Por exemplo, um fundo de cobertura importa dados de ações de um sistema de origem. O preço das ações da Acme Corp muda a 1 de janeiro, mas o fundo só recebe essa atualização a 5 de janeiro. O Bitemporal AUTO CDC permite ao fundo responder a duas questões distintas: qual era o preço real das ações da Acme Corp a 1 de janeiro (horário comercial) e que preço o sistema acreditava quando o fundo tomou decisões de negociação a 3 de janeiro (hora do sistema). A capacidade de distinguir entre estes prazos é útil para auditoria, relatórios regulatórios e tomada de decisões financeiras.

Para permitir o processamento bitemporal, use STORED AS BITEMPORAL (SQL) ou stored_as_scd_type="bitemporal" (Python), use SEQUENCE BY para a coluna de tempo de negócio e use SYSTEM SEQUENCE BY para a coluna de tempo de sistema. A tabela alvo adiciona as colunas __SYSTEM_START_AT e __SYSTEM_END_AT a par das colunas __START_AT e __END_AT do SCD Tipo 2. Para detalhes de sintaxe, consulte AUTO CDC INTO (pipelines) ou create_auto_cdc_flow.

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

Usar várias colunas para sequenciamento

Para sequenciar por múltiplas colunas (por exemplo, um carimbo temporal e um ID para desempatar), use a STRUCT para as combinar. A API ordena inicialmente pelo primeiro campo e, em caso de empate, considera o segundo campo, e assim sucessivamente.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

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

Exemplos AUTO CDC

Os exemplos seguintes demonstram o processamento SCD Tipo 1 e Tipo 2 usando uma fonte de dados de alteração. Os dados de exemplo criam novos registos de utilizador, apagam um registo de utilizador e atualizam os registos de utilizador. No exemplo SCD Tipo 1, as últimas UPDATE operações chegam tarde e são removidas da tabela de destino, demonstrando o manejo de eventos fora de ordem.

Seguem-se os registos de entrada usados nestes exemplos. Estes dados são criados ao executar a consulta na secção Criar dados de exemplo .

Se descomentar a última linha na consulta de geração de dados de exemplo, insere o seguinte registo que especifica truncar a tabela (limpar a tabela) em sequenceNum=3:

Criar dados de exemplo

Execute as seguintes instruções para criar um conjunto de dados de exemplo. Este código não se destina a ser executado como parte de uma definição de pipeline. Executa-o a partir da pasta de exploração do teu 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 as atualizações do Tipo 1 de SCD

O SCD Tipo 1 mantém apenas a versão mais recente de cada registo. O exemplo seguinte extrai dados a partir do feed de alterações criado acima e aplica alterações a um alvo de tabela de streaming. O que são oleodutos? para executar este 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;

Após a execução do exemplo SCD Tipo 1, a tabela alvo contém os seguintes registos:

O utilizador 123 (Isabel) foi eliminado e não aparece. O utilizador 125 (Mercedes) mostra apenas a cidade mais recente (Guadalajara) porque o SCD Tipo 1 sobrespõe valores anteriores. A versão anterior em UPDATE em sequenceNum=5 foi abandonada porque chegou uma atualização posterior em sequenceNum=6.

Após executar o exemplo com o TRUNCATE registo sem comentários, a tabela é limpa em sequenceNum=3. Isto significa que os registos 124 e 126 não estão na tabela, e a tabela alvo final contém apenas o seguinte registo:

Processar atualizações do Tipo 2 de SCD

O SCD Tipo 2 preserva um histórico completo das alterações criando novas linhas para cada versão de um registo, com __START_AT colunas e __END_AT a indicar quando cada versão esteve 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;

Após executar o exemplo SCD Tipo 2, a tabela alvo contém os seguintes registos:

A tabela preserva a história completa. O Utilizador 123 tem duas versões (terminadas na sequência 6 quando eliminadas). Utilizador 125 tem três versões que mostram alterações de cidade. Os registos com __END_AT = null estão atualmente ativos.

Rastreie um subconjunto de colunas com SCD Tipo 2

Por defeito, o SCD Tipo 2 cria uma nova versão sempre que o valor de qualquer coluna muda. Pode especificar um subconjunto de colunas a acompanhar, para que as alterações noutras colunas atualizem a versão atual em vez de gerar um novo registo de histórico.

O exemplo seguinte exclui a city coluna do registo 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 city as alterações não são acompanhadas, as atualizações da cidade sobrescrevem a linha atual em vez de criar uma nova versão. A tabela alvo contém os seguintes registos:

Exemplos de CD AUTOMÁTICO A PARTIR DE SNAPSHOT

As secções seguintes fornecem exemplos da utilização de AUTO CDC FROM SNAPSHOT para processar instantâneos em tabelas alvo SCD Tipo 1 ou Tipo 2. Para contexto sobre quando usar esta API, veja Alterar captura de dados e instantâneos.

Exemplo: Instantâneos de processos usando o tempo de ingestão do pipeline

Utilize esta abordagem quando os snapshots chegarem regularmente e em ordem e pode confiar no timestamp da execução do pipeline para a versionação. Um novo snapshot é importado a cada atualização do pipeline.

Pode ler snapshots de vários tipos de fonte, incluindo tabelas Delta, ficheiros de armazenamento na cloud e ligações JDBC.

Passo 1: Criar dados de exemplo

Crie uma tabela com dados de instantâneo. Execute o seguinte código a partir de um notebook ou do Databricks SQL na explorations pasta 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');

Passo 2: Executar AUTO CDC A PARTIR DO SNAPSHOT

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

Escolha um tipo de fonte para a visualização snapshot (o código de criação do exemplo 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()
  )

Escrever todas as opções no destino

Depois adiciona a tabela alvo 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 primeira execução do pipeline, todos os registos são inseridos como linhas ativas:

Passo 3: Simular um novo snapshot e repetir

Atualize a tabela de origem para simular a chegada de um novo snapshot (execute este código a partir de um caderno ou ficheiro SQL na explorations pasta do seu pipline):

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 snapshot com o anterior e deteta que o utilizador 1 foi eliminado, os utilizadores 2 e 3 foram atualizados, e os utilizadores 4 e 6 foram inseridos. Isto gera um feed de alterações e é usado AUTO CDC para criar a tabela de saída.

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

O Utilizador 1 foi encerrado (apagado). Os utilizadores 2 e 3 têm cada um duas versões que mostram as alterações das suas cidades. Os utilizadores 4 e 6 foram recém-inseridos.

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

Exemplo: Processar snapshots usando funções de versão

Use esta abordagem quando precisar de controlo explícito sobre a ordenação de snapshots. Por exemplo, use esta abordagem quando vários snapshots chegam ao mesmo tempo, ou quando chegam fora de ordem. Escreves uma função que especifica qual snapshot processar a seguir e o seu número de versão. A interface da API processa capturas por ordem crescente de versões.

• Se vários snapshots estiverem armazenados, todos são processados por ordem.
• Se um snapshot chegar fora de ordem (por exemplo, snapshot_3 chegar depois de snapshot_4), é ignorado.
• Se não houver novos snapshots, a função retorna None e não ocorre qualquer processamento.

Passo 1: Preparar ficheiros de instantâneo

Crie ficheiros CSV contendo dados snapshot e adicione-os a um volume ou local de armazenamento na cloud. Nomeie os ficheiros cronologicamente (por exemplo, snapshot_1.csv, snapshot_2.csv).

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

snapshot_1.csv:

snapshot_2.csv:

Passo 2: Execute AUTO CDC A PARTIR DO SNAPSHOT com uma função de versão

Crie um novo notebook e cole o seguinte código do 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 de snapshot_1.csv, a tabela de destino contém os seguintes registos:

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

Passo 3: Adicionar novos snapshots

Adicione um novo ficheiro CSV à localização de armazenamento com dados modificados (por exemplo, valores de cidade alterados, novas linhas ou linhas removidas). Depois executa novamente o pipeline para processar o novo snapshot.

Exemplos bitemporais de AUTO CDC

O exemplo seguinte cria uma tabela alvo bitemporal a partir de um pequeno conjunto de eventos CDC sintéticos. A bt coluna tem a hora de negócio e a st coluna tem 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;

Os passos seguintes explicam como uma tabela bitemporal regista inserções, atualizações, atualizações fora de ordem e eliminações para uma única empresa. A coluna de sequenciação gera as __START_AT colunas e __END_AT (tempo de negócio), e a coluna de sequenciação do sistema gera as __SYSTEM_START_AT colunas e __SYSTEM_END_AT (tempo do sistema):

O sistema lida com eventos que chegam em qualquer ordem em ambas as linhas temporais. Quando um evento chega com um tempo de negócio ou tempo de sistema anterior ao dos eventos já processados, o sistema corrige o histórico afetado em vez de acrescentar apenas no fim.

Passo 1: Inserir

A Empresa A é adicionada a 18/07/2025 às 10:01:00 (hora comercial), mas só é ingerida às 10:05:00 (hora do sistema).

Entrada:

Output:

O XFv1 é válido a partir das 10:01:00 sem fim conhecido. O sistema soube deste facto às 10:05:00, sem fim conhecido.

Passo 2: Atualização

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

Entrada:

Output:

XFv1 era considerado válido a partir das 10:01:00 sem fim conhecido, e o sistema manteve essa crença das 10:05:00 até às 12:20:00. Sabe-se agora que XFv1 é válido apenas até às 12:15:43, com um histórico corrigido em vigor a partir da hora do sistema 12:20:00, sem data de fim conhecida. XFv2 é válido a partir de 12:15:43, sem hora de fim conhecida, e foi registado à hora do sistema 12:20:00.

Passo 3: Atualização fora de ordem

Chega uma atualização fora de ordem indicando que a Empresa A foi na verdade atualizada a 18/07/2025 às 12:05:00 (hora comercial), mas só é ingerida às 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 tempo de negócio histórico e preserva tanto aquilo que o sistema considerava antes da chegada da atualização fora de ordem como o histórico corrigido.

Entrada:

Output:

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

Passo 4: Eliminar

A Empresa A é eliminada a 18/07/2025 às 12:30:00, e o sistema consome o evento às 12:30:00. Como uma operação de eliminação representa o fim da existência comercial da entidade, o sistema não cria nenhuma linha de substituição. O XFv2 aparece em duas linhas, preservando um registo completo de auditoria tanto de quando a empresa deixou de existir como de quando o sistema tomou conhecimento da eliminação.

Entrada:

Output:

XFv2 era válido a partir das 12:15:43 sem fim conhecido, e o sistema mantinha essa crença das 12:20:00 às 12:30:00. Depois de a eliminação ser assimilada, sabe-se que o XFv2 só é válido até às 12:30:00, sendo o histórico corrigido efetivo a partir da hora do sistema 12:30:00.

Limitações

• A coluna de sequenciação deve ser um tipo de dado ordenável. NULL Os valores de sequenciação não são suportados.
• AUTO CDC FROM SNAPSHOT é suportado apenas na interface do pipeline Python; a interface SQL não é suportada.
• Para obter dados em fluxo a partir do destino de um processo AUTO CDC, leia o respetivo fluxo de alterações. Para mais detalhes, consulte Ler um feed de dados de alteração a partir de uma tabela alvo AUTO CDC.

Recursos adicionais

• Captação de Dados de Alterações e snapshots: Aprenda sobre conceitos de CDC, snapshots e tipos de SCD.
• Replicar uma tabela RDBMS externa usando AUTO CDC: Aprenda a realizar a hidratação inicial com um once fluxo e depois continue a processar as alterações.
• Tópicos avançados do AUTO CDC: Aprenda sobre operações de alteração nos alvos do 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