Tópicos avançados do AUTO CDC

Além das APIs básicas AUTO CDC e AUTO CDC FROM SNAPSHOT, você pode executar DML em tabelas de destino, ler feeds de dados de mudança de destinos de CDC, monitorar métricas de processamento, aplicar atualizações parciais e acompanhar alterações com armazenamento bitemporal. Para obter uma introdução às APIs AUTO CDC, consulte as APIs AUTO CDC: Simplifique a captura de dados de alterações com pipelines.

Adicionar, alterar ou excluir dados em uma tabela de streaming de destino

Se o pipeline publicar tabelas no Catálogo do Unity, você poderá usar instruções DML linguagem de manipulação de dados, incluindo instruções de inserção, atualização, exclusão e mesclagem, para modificar as tabelas de streaming de destino criadas por instruções AUTO CDC ... INTO.

Como o pipeline usa uma coluna específica SEQUENCE BY e propaga valores de sequenciamento apropriados para as colunas __START_AT e __END_AT da tabela de destino (para SCD Tipo 2), você deve garantir que as instruções DML usem valores válidos para essas colunas para manter a ordenação correta dos registros. Veja como o AUTO CDC funciona.

Para obter mais informações sobre como usar instruções DML com tabelas de streaming, consulte Adicionar, alterar ou excluir dados em uma tabela de streaming.

O exemplo a seguir insere um registro ativo com uma sequência inicial de 5:

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

CREATE VIEW my_employees_view AS
SELECT
  *,
  __START_AT AS valid_from,
  __END_AT AS valid_to
FROM my_scd2_target_table;

Ler um fluxo de dados de alteração de uma tabela de destino AUTO CDC

No Databricks Runtime 15.2 e superior, você pode ler um feed de dados de alterações de uma tabela de streaming que é o destino de consultas AUTO CDC ou AUTO CDC FROM SNAPSHOT da mesma forma que lê um feed de dados de alterações de outras tabelas Delta. Para ler o feed de dados de alteração de uma tabela de streaming de destino, são necessários os seguintes elementos:

• A tabela de streaming de destino deve ser publicada no Catálogo do Unity. Consulte Usar o Catálogo do Unity com pipelines.
• Para ler o feed de dados de alteração da tabela de streaming de destino, você deve usar o Databricks Runtime 15.2 ou superior. Para ler o feed de dados das alterações em um pipeline diferente, é necessário configurar o pipeline para utilizar o Databricks Runtime 15.2 ou versão posterior.

Você lê o fluxo de dados de alteração de uma tabela de streaming de destino criada em um pipeline do Lakeflow da mesma forma que você lê um fluxo de dados de alteração de outras tabelas Delta. Para saber mais sobre como usar a funcionalidade de feed de dados de alterações Delta, incluindo exemplos em Python e SQL, consulte Usar o feed de dados de alterações no Azure Databricks.

Leia um feed de dados de mudança a partir de uma visualização materializada

Você pode ler um feed de dados de mudança a partir de uma visualização materializada criada em um pipeline Lakeflow ou no SQL do Databricks. Use isso para replicar alterações materializadas na visualização em destinos fora do Azure Databricks, ou para manter um histórico das alterações materializadas na visualização para auditoria e relatórios.

Visualizações materializadas usam fluxo automático de dados de alteração, então você não ativa o próprio feed de dados de alteração. Em vez disso, ative o feed de dados de alterações em cada exibição materializada na qual você precisar dele, desde que os seguintes requisitos sejam atendidos. Consulte o feed automático de alterações de dados.

• Para ler o feed de dados de alterações, você deve usar o Databricks Runtime 18 LTS ou superior, na computação clássica, na computação serverless ou no Databricks SQL.

• A visualização materializada, o pipeline que a cria ou o pipeline que a lê devem usar o PREVIEW canal.

• A visualização materializada deve ter o rastreamento de linhas ativado. As visualizações materializadas na computação sem servidor têm o rastreamento de linhas ativado por padrão. Consulte o rastreamento de linhas no Azure Databricks. Para verificar se o rastreamento de linhas está ativado em uma visualização materializada, execute:

  SHOW TBLPROPERTIES my_mv ('delta.enableRowTracking');
  

• Para ler o fluxo de dados de alterações de uma visualização materializada, ative o sinalizador de metadados externos no pipeline ou na visualização materializada. Para instruções, veja Como habilitar o acesso para um conjunto de dados.

Você lê o feed de dados de alterações de uma visualização materializada da mesma forma que em outras tabelas Delta, usando a função table_changes(), uma leitura em streaming ou a opção readChangeFeed. Para sintaxe e exemplos em SQL e Python, veja Usar feed de dados de alteração no Azure Databricks.

Você pode ler um fluxo de dados de mudança de vista materializado dentro de uma visualização materializada SQL do Databricks ou tabela de streaming:

CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT * FROM STREAM my_mv WITH (readChangeFeed=true)

Limitações

Além das limitações do feed automático de dados de alteração, o seguinte se aplica quando você lê um feed de dados de alteração a partir de uma visualização materializada:

• O feed de dados de mudança inclui linhas inalteradas quando a visualização materializada é totalmente reescrita, e ele não consolida múltiplas atualizações da mesma linha em um único evento. Para filtrar esses dados, agregue o feed de dados de alteração agrupando em todas as colunas para encontrar inserções e excluções que compartilhem os mesmos valores de linha.
• Somente o Azure Databricks pode consultar o feed de dados de alteração para obter uma visualização materializada. Clientes externos de Delta Lake e Iceberg não podem.
• Em pipelines do Lakeflow, você pode ler o feed de dados de alterações de uma exibição materializada somente a partir de outro pipeline, e esse pipeline deve usar o canal PREVIEW. A leitura do feed de dados de alterações de uma visualização materializada no mesmo pipeline que a cria não é suportada.
• Você não pode criar um índice de busca vetorial a partir de uma visualização materializada.

Obter dados sobre registros processados por uma consulta CDC em pipelines

As seguintes métricas são capturadas por AUTO CDC consultas:

• num_upserted_rows: o número de linhas de saída para as quais foi executado upsert no conjunto de dados durante uma atualização.
• num_deleted_rows: o número de linhas de saída existentes excluídas do conjunto de dados durante uma atualização.

A métrica num_output_rows, saída para fluxos não CDA, não é capturada para consultas AUTO CDC.

Aplicar atualizações parciais

Quando uma origem envia apenas as colunas que foram alteradas, AUTO CDC deve distinguir entre uma coluna ausente de um registro de alteração, o que deve deixar o valor de destino inalterado e uma coluna definida explicitamente como null, que deve substituir o valor de destino com null. Por padrão, IGNORE NULL UPDATES trata cada null como um marcador "não atualizar", portanto, não pode aplicar um null explícito. Para resolver essa ambiguidade, escolha um dos três métodos a seguir:

COLUMNS TO UPDATE não pode ser combinado com IGNORE NULL UPDATES, e não tem suporte para tabelas bitemporais.

Como regra prática, escolha COLUMNS TO UPDATE quando o produtor souber quais colunas foram alteradas em cada registro e puder armazenar essa informação em uma coluna na origem, como quando vários produtores gravam na mesma fonte ou o conjunto de colunas atualizáveis aumenta ao longo do tempo. Escolha IGNORE NULL UPDATES ON quando o proprietário do pipeline souber o conjunto fixo de colunas atualizáveis com antecedência e preferir controlá-las no código do pipeline.

O exemplo a seguir usa uma coluna de origem nomeada columnsToUpdate para controlar quais colunas cada altera atualizações de registro, incluindo colunas definidas explicitamente como null:

Python

from pyspark import pipelines as dp

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "sequenceNum",
  stored_as_scd_type = 1,
  columns_to_update = "columnsToUpdate"
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  target
FROM
  stream(cdc_source)
KEYS
  (id)
SEQUENCE BY
  sequenceNum
STORED AS
  SCD TYPE 1
COLUMNS TO UPDATE
  columnsToUpdate;

Para obter a referência completa do parâmetro, consulte AUTO CDC INTO (pipelines) e create_auto_cdc_flow.

Bitemporal AUTO CDC

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.

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;

A sequência de alterações a seguir mostra como uma tabela bitemporal registra uma inserção, uma atualização, uma atualização fora de ordem e uma exclusão 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.

Alteração 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.

Alteração 2: Atualização

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.

Alteração 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 (horário comercial), 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.

Alteração 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.

Quais são os objetos de dados usados para o processamento de CDC em um pipeline?

Quando você declara a tabela de destino no metastore do Hive, duas estruturas de dados são criadas:

• Uma exibição usando o nome atribuído à tabela de destino.
• Uma tabela de backup interna usada pelo pipeline para gerenciar o processamento de CDC. Esta tabela é nomeada anexando-se __apply_changes_storage_ ao nome da tabela de destino.

Por exemplo, se você declarar uma tabela de destino chamada dp_cdc_target, verá uma exibição nomeada dp_cdc_target e uma tabela nomeada __apply_changes_storage_dp_cdc_target no metastore. Consulte a visualização para acessar os dados processados. Não modifique a tabela de backup diretamente.