Tópicos Avançados do AUTO CDC

Para além das APIs básicas AUTO CDC e AUTO CDC FROM SNAPSHOT, pode executar DML em tabelas de destino, ler fluxos de dados de alterações de destinos CDC, monitorizar métricas de processamento, aplicar atualizações parciais e rastrear alterações com armazenamento bitemporal. Para uma introdução às AUTO CDC APIs, veja As APIs AUTO CDC: Simplificar 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 Unity Catalog, você poderá usar instruções DML (linguagem de manipulação de dados ), incluindo instruções insert, update, delete e merge, para modificar as tabelas de streaming de destino criadas pelas AUTO CDC ... INTO instruções.

Como o pipeline utiliza uma coluna especificada SEQUENCE BY e propaga valores de sequenciação apropriados para as __START_AT colunas e __END_AT da tabela alvo (para SCD Tipo 2), deve garantir que as instruções DML usam valores válidos para estas colunas para manter a ordem correta dos registos. Veja como funciona o AUTO CDC.

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 do AUTO CDC

No Databricks Runtime 15.2 e superior, pode ler um fluxo de dados de alteração de uma tabela de fluxo contínuo que é o alvo de consultas AUTO CDC ou AUTO CDC FROM SNAPSHOT, da mesma forma que lê um fluxo de dados de alteração de outras tabelas Delta. Para ler a alimentação de dados de alteração de uma tabela de transmissão alvo, são necessários os seguintes requisitos:

• A tabela de streaming de destino deve ser publicada no Catálogo Unity. Consulte Utilizar o catálogo 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 de alteração num pipeline diferente, é necessário que o pipeline seja configurado para usar o Databricks Runtime 15.2 ou superior.

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

Leia um fluxo de dados de alterações de uma vista materializada

Pode ler um fluxo de dados de alterações de uma vista materializada criada num pipeline Lakeflow ou no Databricks SQL. Use isto para replicar alterações materializadas na vista para destinos fora do Azure Databricks, ou para manter um histórico das alterações materializadas na visualização para auditoria e reporte.

As visualizações materializadas utilizam automaticamente o feed de dados alterados, pelo que não é necessário ativar o feed de dados alterados em si. Em vez disso, ativa o feed de dados de alteração em cada vista materializada onde precisa que seja implementado, cumprindo os seguintes requisitos. Consulte alimentação de dados de alterações automática.

• Para ler o feed de dados de alterações, tem de usar o Databricks Runtime 18 LTS ou superior, em classic compute, serverless compute ou Databricks SQL.

• A visualização materializada, a canalização que a cria ou a canalização que a lê tem de usar o canal PREVIEW.

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

  SHOW TBLPROPERTIES my_mv ('delta.enableRowTracking');
  

• Para ler o fluxo de dados de alterações a partir de uma vista materializada, ative a opção de metadados externos na pipeline ou na vista materializada. Para instruções, veja Como permitir o acesso a um conjunto de dados.

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

Pode ler um fluxo de dados de alterações de uma vista materializada a partir de uma vista materializada do Databricks SQL ou de uma tabela de fluxo contínuo:

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

Limitations

Para além das limitações automáticas do feed de dados de alterações, aplica-se o seguinte ao ler um feed de dados de alterações de uma vista materializada:

• O feed de dados de alteração inclui linhas inalteradas quando a visualização materializada é totalmente reescrita, e não consolida múltiplas atualizações da mesma linha num único evento. Para os filtrar, agregue o fluxo de dados de alterações, agrupando por todas as colunas, para encontrar inserções e eliminações que partilhem os mesmos valores de registo.
• Apenas o Azure Databricks pode consultar o feed de dados de alteração para obter uma visualização materializada. Clientes externos do Delta Lake e do Iceberg não podem.
• Nos pipelines Lakeflow, só pode ler o feed de dados de alterações de uma vista materializada a partir de outro pipeline, e esse pipeline tem de usar o canal PREVIEW. Não é suportada a leitura do feed de dados de alterações de uma vista materializada no mesmo pipeline que a cria.
• Não se pode criar um índice de pesquisa vetorial a partir de uma vista materializada.

Obter dados sobre registos processados por uma consulta de 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 inseridas 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, resultado para fluxos não-CDC, não é capturada nas consultas AUTO CDC.

Aplicar atualizações parciais

Quando uma fonte envia apenas as colunas que mudaram, AUTO CDC deve distinguir entre uma coluna ausente de um registo de alteração, que deve deixar o valor alvo inalterado, e uma coluna explicitamente definida para null, que deve sobrescrever o valor alvo por null. Por predefinição, IGNORE NULL UPDATES trata todos os null como marcadores de "não atualizar", pelo que não pode aplicar um null explícito. Para resolver esta ambiguidade, escolha um dos seguintes três métodos:

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

Como regra geral, escolha COLUMNS TO UPDATE quando o produtor sabe quais as colunas alteradas em cada registo e pode transportar essa informação numa coluna de origem, como quando vários produtores escrevem para a mesma fonte ou quando o conjunto de colunas atualizáveis cresce ao longo do tempo. Escolha IGNORE NULL UPDATES ON quando o proprietário do pipeline conhece antecipadamente o conjunto fixo de colunas atualizáveis e prefere controlá-las no código do pipeline.

O exemplo seguinte utiliza uma coluna de origem com o nome columnsToUpdate para controlar quais as colunas que cada registo de alteração atualiza, 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 a referência completa dos parâmetros, veja AUTO CDC INTO (pipelines) e create_auto_cdc_flow.

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.

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;

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

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

Saída:

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.

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

Saída:

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.

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

Saída:

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.

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

Saída:

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.

Quais objetos de dados são usados para processamento CDC num pipeline?

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

• Um modo de exibição usando o nome atribuído à tabela de destino.
• Uma tabela de suporte interna usada pelo pipeline para gerenciar o processamento CDC. Esta tabela é nomeada por preceder __apply_changes_storage_ o nome da tabela de destino.

Por exemplo, se declarar uma tabela alvo chamada dp_cdc_target, verá uma vista nomeada dp_cdc_target e uma tabela nomeada __apply_changes_storage_dp_cdc_target na metastore. Consulta a visualização para aceder aos dados processados. Não modifiquem diretamente a mesa de apoio.