Argomenti avanzati di AUTO CDC

Oltre alle API di base AUTO CDC e AUTO CDC FROM SNAPSHOT, è possibile eseguire operazioni DML sulle tabelle di destinazione, leggere i feed dei dati di modifica dai target CDC, monitorare le metriche di elaborazione, applicare aggiornamenti parziali e tenere traccia delle modifiche con l'archiviazione bitemporale. Per un'introduzione alle AUTO CDC API, vedere Le API AUTO CDC: semplificare la cattura dei dati delle modifiche con le pipeline.

Aggiungere, modificare o eliminare dati in una tabella di streaming di destinazione

Se la pipeline pubblica tabelle in Unity Catalog, è possibile usare istruzioni DML ( Data Manipulation Language ), incluse le istruzioni insert, update, delete e merge, per modificare le tabelle di streaming di destinazione create dalle AUTO CDC ... INTO istruzioni .

Poiché la pipeline usa una colonna specificata SEQUENCE BY e propaga i valori di sequenziazione appropriati alle __START_AT colonne e __END_AT della tabella di destinazione (per scD Type 2), è necessario assicurarsi che le istruzioni DML usino valori validi per queste colonne per mantenere l'ordinamento corretto dei record. Vedere Funzionamento di AUTO CDC.

Per altre informazioni sull'uso di istruzioni DML con tabelle di streaming, vedere Aggiungere, modificare o eliminare dati in una tabella di streaming.

Nell'esempio seguente viene inserito un record attivo con una sequenza iniziale pari a 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;

Leggere un feed di dati di modifica da una tabella di destinazione AUTO CDC

In Databricks Runtime 15.2 e versioni successive, è possibile leggere un feed di dati delle modifiche da una tabella di streaming che è la destinazione delle query AUTO CDC o AUTO CDC FROM SNAPSHOT nello stesso modo in cui si legge un feed di dati delle modifiche da altre tabelle Delta. Per leggere il feed di dati delle modifiche da una tabella di streaming di destinazione, è necessario quanto segue:

• La tabella di streaming di destinazione deve essere pubblicata in Unity Catalog. Vedere Usare il catalogo Unity con le pipeline.
• Per leggere il feed di dati delle modifiche dalla tabella di streaming di destinazione, è necessario usare Databricks Runtime 15.2 o versione successiva. Per leggere il feed di dati delle modifiche in una pipeline diversa, la pipeline deve essere configurata per l'uso di Databricks Runtime 15.2 o versione successiva.

Il feed di dati delle modifiche viene letto da una tabella di streaming di destinazione creata in una pipeline Lakeflow allo stesso modo in cui si legge un feed di dati delle modifiche da altre tabelle Delta. Per altre informazioni sull'uso della funzionalità feed di dati delle modifiche Delta, inclusi esempi in Python e SQL, vedere Usare il feed di dati delle modifiche in Azure Databricks.

Leggi un feed di dati di cambiamento da una vista materializzata

Puoi leggere un feed di dati di cambiamento da una vista materializzata creata in una pipeline Lakeflow o in Databricks SQL. Usa questo per replicare le modifiche apportate alla vista materializzata in destinazioni esterne ad Azure Databricks o per mantenere una cronologia delle modifiche della vista materializzata a fini di controllo e creazione di report.

Le visualizzazioni materializzate utilizzano il feed automatico dei dati di modifica, quindi non si attiva il feed di dati di modifica stesso. Invece, attivi il feed di modifica dei dati su ogni visualizzazione materializzata su cui hai bisogno soddisfando i seguenti requisiti. Vedere Feed automatico dei dati delle modifiche.

• Per leggere il feed dei dati di modifica, è necessario utilizzare Databricks Runtime 18 LTS o versione successiva, su compute classico, compute serverless o Databricks SQL.

• La vista materializzata, la pipeline che la crea o la pipeline che la legge devono utilizzare il canale PREVIEW.

• La visualizzazione materializzata deve avere il tracciamento delle righe abilitato. Le visualizzazioni materializzate nel calcolo serverless hanno il tracciamento delle righe abilitato per impostazione predefinita. Vedi Tracciamento delle righe in Azure Databricks. Per verificare se il tracciamento delle righe è abilitato su una visualizzazione materializzata, esegui:

  SHOW TBLPROPERTIES my_mv ('delta.enableRowTracking');
  

• Per leggere il feed di dati di modifica da una visualizzazione materializzata, abilita il flag di metadati esterno sulla pipeline o sulla visualizzazione materializzata. Per le istruzioni, vedi Come abilitare l'accesso a un dataset.

Leggi il feed dei dati di modifica da una vista materializzata allo stesso modo che da altre tabelle Delta, usando la funzione table_changes(), una lettura in streaming o l'opzione readChangeFeed. Per sintassi ed esempi in SQL e Python, vedi Usa il feed di modifica dei dati su Azure Databricks.

Puoi leggere un feed di dati di cambiamento di visualizzazione materializzato dall'interno di una vista materializzata SQL di Databricks o di una tabella di streaming:

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

Limitations

Oltre alle limitazioni del feed automatico dei dati di modifica, si applica quanto segue quando si legge un feed dei dati di modifica da una vista materializzata:

• Il flusso di dati di modifica include righe inalterate quando la vista materializzata è completamente riscritta, e non consolida più aggiornamenti della stessa riga in un unico evento. Per filtrare questi, aggrega il feed dei dati di modifica raggruppandolo su tutte le colonne per trovare inserti ed eliminazioni che condividono gli stessi valori delle righe.
• Solo Azure Databricks può interrogare il feed dei dati di modifica per una visualizzazione materializzata. I clienti esterni di Delta Lake e Iceberg non possono.
• All'interno delle pipeline Lakeflow, puoi leggere un feed di dati di cambiamento di vista materializzato solo da una pipeline diversa, e quella pipeline deve utilizzare il PREVIEW canale. La lettura del feed di dati di cambiamento di una vista materializzata nella stessa pipeline che la crea non è supportata.
• Non puoi creare un indice di ricerca vettoriale da una vista materializzata.

Ottieni dati sui record elaborati da una query CDC delle pipeline

Le metriche seguenti vengono acquisite dalle AUTO CDC query:

• num_upserted_rows: numero di righe di output inserite nel set di dati durante un aggiornamento.
• num_deleted_rows: numero di righe di output esistenti eliminate dal set di dati durante un aggiornamento.

La metrica num_output_rows, l'output per i flussi non CDC, non viene acquisita per le query AUTO CDC.

Applicare aggiornamenti parziali

Quando un'origine invia solo le colonne modificate, AUTO CDC deve distinguere una colonna assente da un record di modifica, che deve lasciare invariato il valore di destinazione e una colonna impostata in modo esplicito su null, che deve sovrascrivere il valore di destinazione con null. Per impostazione predefinita, IGNORE NULL UPDATES considera ogni null elemento come marcatore "non aggiornare", quindi non può applicare un oggetto esplicito null. Per risolvere questa ambiguità, scegliere uno dei tre metodi seguenti:

COLUMNS TO UPDATE non può essere combinato con IGNORE NULL UPDATESe non è supportato per le tabelle bitemporal.

Come regola generale, scegliere COLUMNS TO UPDATE quando il produttore sa quali colonne sono cambiate in ogni record e può riportare queste informazioni in una colonna di origine, ad esempio quando più produttori scrivono nella stessa origine dati o l’insieme delle colonne aggiornabili aumenta nel tempo. Scegliere IGNORE NULL UPDATES ON quando il proprietario della pipeline conosce in anticipo il set fisso di colonne aggiornabili e preferisce controllarle nel codice della pipeline.

L'esempio seguente utilizza una colonna di origine denominata columnsToUpdate per controllare quali colonne vengono aggiornate da ogni record di modifica, comprese le colonne impostate esplicitamente su 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;

Per il riferimento completo dei parametri, vedere AUTO CDC INTO (pipelines) e create_auto_cdc_flow.

Bitemporale AUTO CDC

ScD Type 1 e Type 2 sono unitemporal: tengono traccia delle modifiche in una singola dimensione temporale. Bitemporal estende la cronologia scD Type 2 per tenere traccia delle modifiche tra due dimensioni temporali e distinguere tra due prospettive:

• Ora aziendale: quando l'evento è effettivamente accaduto.
• Ora di sistema: quando il sistema ha registrato o inserito l'evento.

Analogamente al tipo SCD 2, bitemporal mantiene una cronologia completa dei record. Aggiunge una seconda sequenza temporale in modo da poter ricostruire sia i dati mostrati che quello che il sistema ha creduto in qualsiasi momento del passato.

Ad esempio, un hedge fund inserisce i dati delle scorte da un sistema di origine. Il prezzo azionario di Acme Corp cambia il 1° gennaio, ma il fondo non inserisce tale aggiornamento fino al 5 gennaio. Bitemporal AUTO CDC consente al fondo di rispondere a due domande distinte: quale prezzo effettivo delle azioni di Acme Corp era il 1° gennaio (ora aziendale) e quale prezzo il sistema ha creduto quando il fondo ha preso decisioni di trading il 3 gennaio (ora di sistema). La possibilità di distinguere tra queste sequenze temporali è utile per il controllo, la creazione di report normativi e il processo decisionale finanziario.

Per abilitare l'elaborazione bitemporal, impostare STORED AS BITEMPORAL (SQL) o stored_as_scd_type="bitemporal" (Python), usare SEQUENCE BY per la colonna ora aziendale e usare SYSTEM SEQUENCE BY per la colonna ora di sistema. La tabella di destinazione aggiunge le colonne __SYSTEM_START_AT e __SYSTEM_END_AT oltre alle colonne __START_AT e __END_AT di tipo SCD 2. Per informazioni dettagliate sulla sintassi, vedere AUTO CDC INTO (pipeline) o create_auto_cdc_flow.

Esempi di AUTO CDC bitemporale

L'esempio seguente crea una tabella di destinazione bitemporale a partire da un piccolo insieme di eventi CDC sintetici. La colonna bt contiene l'ora aziendale e la colonna st contiene l'ora di 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;

La sequenza di modifiche seguente mostra come una tabella bitemporal registra un inserimento, un aggiornamento, un aggiornamento non ordinato e un'eliminazione per una singola società. La colonna di sequenziazione genera le __START_AT colonne e __END_AT (ora aziendale) e la colonna di sequenziazione del sistema genera le __SYSTEM_START_AT colonne e __SYSTEM_END_AT (ora di sistema):

Il sistema gestisce gli eventi che arrivano in qualsiasi ordine in entrambe le sequenze temporali. Quando un evento arriva con un tempo operativo o un'ora di sistema anteriore a quello degli eventi già elaborati, il sistema corregge la cronologia interessata anziché limitarsi ad aggiungerla alla fine.

Modifica 1: Inserisci

La Società A viene aggiunta alle ore 18/07/2025 10:01:00 (ora aziendale), ma non viene inserita fino alle 10:05:00 (ora di sistema).

Inserimento

Risultato:

XFv1 è valido a partire dalle 10:01:00 senza fine nota. Il sistema ha rilevato questo fatto all’ora di sistema 10:05:00, senza un termine noto.

Modifica 2: Aggiornamento

La Società A viene aggiornata il 18/07/2025 alle 12:15:43 (ora aziendale) e il sistema elabora l'evento alle 12:20:00 (ora di sistema). Il sistema mantiene sia ciò che credeva prima che l'aggiornamento fosse noto e la cronologia aziendale corretta dopo l'inserimento dell'aggiornamento.

Inserimento

Risultato:

XFv1 è stato creduto valido dalle 10:01:00 senza fine nota e il sistema ha mantenuto tale convinzione dalle 10:05:00 alle 12:20:00. XFv1 è ora noto come valido solo fino alle 12:15:43, una cronologia corretta valida a partire dall'ora di sistema 12:20:00 senza fine nota. XFv2 è valido a partire dalle 12:15:43 senza una data di fine nota ed è stato acquisito all'ora di sistema delle 12:20:00.

Modifica 3: Aggiornamento non ordinato

Arriva un aggiornamento fuori sequenza che indica che l'Azienda A è stata effettivamente aggiornata il 18/07/2025 alle 12:05:00 (orario di business), ma l'aggiornamento non viene acquisito fino alle 12:25:00 (ora di sistema). Quando un aggiornamento arriva successivamente nel tempo di sistema, ma con un tempo di business precedente, il sistema corregge lo storico del tempo di business e preserva sia ciò che riteneva prima dell'aggiornamento fuori ordine sia la cronologia corretta.

Inserimento

Risultato:

XFv1 si riteneva valido dalle 10:01:00 alle 12:15:43 e tale valutazione è ora valida nel tempo di sistema fino alle 12:25:00. Il nuovo aggiornamento corregge la validità di business di XFv1, facendola terminare alle 12:05:00, con una cronologia corretta efficace a partire dall'ora di sistema 12:25:00. XFv3 è ora noto come valido dalle 12:05:00 alle 12:15:43, una convinzione valida nell'ora di sistema dalle 12:25:00 senza fine nota.

Modifica 4: Eliminazione

La società A viene eliminata il 18/7/2025 alle 12:30:00, e il sistema consuma l'evento alle 12:30:00. Poiché un'operazione di eliminazione rappresenta la fine dell'esistenza aziendale dell'entità, il sistema non crea alcuna riga di sostituzione. XFv2 viene visualizzato in due righe, preservando una traccia di audit completa sia del momento in cui l'azienda ha cessato di esistere sia di quello in cui il sistema ha rilevato l'eliminazione.

Inserimento

Risultato:

XFv2 è valido dalle 12:15:43 senza fine nota e il sistema ha mantenuto tale convinzione dalle 12:20:00 alle 12:30:00. Dopo l'inserimento dell'eliminazione, XFv2 è noto come valido solo fino alle 12:30:00, una cronologia corretta valida a partire dall'ora di sistema 12:30:00.

Quali oggetti dati vengono usati per l'elaborazione CDC in una pipeline?

Quando si dichiara la tabella di destinazione nel metastore Hive, vengono create due strutture di dati:

• Visualizzazione che usa il nome assegnato alla tabella di destinazione.
• Tabella di backup interna usata dalla pipeline per gestire l'elaborazione CDC. Questa tabella viene denominata anteponendo __apply_changes_storage_ al nome della tabella di destinazione.

Ad esempio, se si dichiara una tabella di destinazione denominata dp_cdc_target, nel metastore viene visualizzata una vista denominata dp_cdc_target e una tabella denominata __apply_changes_storage_dp_cdc_target. Eseguire una query sulla vista per accedere ai dati elaborati. Non modificare direttamente la tabella di backup.