API AUTO CDC: semplificano la cattura delle modifiche ai dati con le pipeline

Le pipeline di Lakeflow semplificano l'acquisizione dei dati modificati (CDC) con le API AUTO CDC e AUTO CDC FROM SNAPSHOT. Queste API automatizzano la complessità del calcolo delle dimensioni a modifica lenta (SCD) di tipo 1 e tipo 2 sia da un feed CDC che da snapshot di database. L'API AUTO CDC supporta anche il rilevamento bitemporale, che registra le modifiche tra due dimensioni temporali (Beta). Per saperne di più su SCD Type 1 e Type 2, vedere Acquisizione dei dati delle modifiche e snapshot. Per altre informazioni sul rilevamento bitemporale, vedere Bitemporal AUTO CDC.

Annotazioni

Le AUTO CDC API sostituiscono le APPLY CHANGES API e hanno la stessa sintassi. Le APPLY CHANGES API sono ancora disponibili, ma Databricks consiglia di usare le AUTO CDC API sul posto.

L'API usata dipende dall'origine dei dati delle modifiche:

  • AUTO CDC: usare questa opzione quando il database di origine dispone di un feed CDC abilitato. AUTO CDC elabora le modifiche da un feed di dati di modifica (CDF). È supportato sia nelle interfacce SQL della pipeline che in Python.
  • AUTO CDC FROM SNAPSHOT: usare questa opzione quando CDC non è abilitato nel database di origine e sono disponibili solo gli snapshot. Questa API confronta gli snapshot per determinare le modifiche e quindi elaborarle. È supportato sia nelle interfacce SQL della pipeline che in Python.

Per i target SCD di tipo 1, si può anche combinare un AUTO CDC FROM SNAPSHOT flusso con uno o più AUTO CDC flussi. Usa questo schema quando una sorgente fornisce snapshot autorevoli periodici e un feed CDC a latenza inferiore tra gli snapshot.

Entrambe le API supportano l'aggiornamento delle tabelle tramite scD Type 1 e Type 2:

  • Usare scD Type 1 per aggiornare direttamente i record. La cronologia non viene mantenuta per i registri aggiornati.
  • Usare SCD Type 2 per conservare una cronologia dei record, su tutti gli aggiornamenti o sugli aggiornamenti di un set specificato di colonne.

Solo per AUTO CDC, è anche possibile usare l'archiviazione bitemporale, che estende la cronologia di SCD Type 2 per tenere traccia delle modifiche lungo due dimensioni temporali: tempo di business e tempo di sistema. Bitemporal è in Beta. Vedere Bitemporal AUTO CDC.

AUTO CDC supporta anche aggiornamenti parziali, in cui un record di modifiche aggiorna solo un subset di colonne. Vedere Applicare aggiornamenti parziali.

Le AUTO CDC API non sono supportate dalle pipeline dichiarative di Apache Spark.

Per la sintassi e altri riferimenti, vedere AUTO CDC INTO (pipeline),create_auto_cdc_flow e create_auto_cdc_from_snapshot_flow.

Annotazioni

Usa queste API per aggiornare le tabelle nelle pipeline in base ai cambiamenti nei dati sorgente. Per informazioni su come registrare ed eseguire query sulle informazioni sulle modifiche a livello di riga per le tabelle Delta, vedere Usare il feed di dati delle modifiche in Azure Databricks.

Requisiti

Per utilizzare le API CDC, la pipeline deve essere configurata per usare le pipeline Lakeflow serverless o le pipeline Lakeflow Pro o Advancededizioni.

Funzionamento di AUTO CDC

Per eseguire l'elaborazione CDC con AUTO CDC, creare una tabella di streaming e quindi usare l'istruzione AUTO CDC ... INTO in SQL o la create_auto_cdc_flow() funzione in Python per specificare l'origine, le chiavi e la sequenziazione per il feed di modifiche. Per una spiegazione del funzionamento della sequenziazione e della logica SCD, vedere Acquisizione dati modificati e istantanee. Vedere gli esempi di AUTO CDC.

Per l'idratazione iniziale da un'origine con un change feed, utilizzare AUTO CDC in un flusso once e quindi continuare a elaborare il change feed. Vedere Replicare una tabella RDBMS esterna usando AUTO CDC.

Per informazioni dettagliate sulla sintassi, vedere AUTO CDC INTO (pipeline) o create_auto_cdc_flow.

Funzionamento di AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT determina le modifiche nei dati di origine confrontando gli snapshot in ordine. È possibile leggere gli snapshot da una tabella Delta, da file di archiviazione cloud o da JDBC direttamente. È supportato sia nelle interfacce SQL della pipeline che in Python.

Per eseguire l'elaborazione CDC con AUTO CDC FROM SNAPSHOT, crea una tabella di streaming e poi definisci il flow:

  • In SQL, si utilizza la forma AUTO CDC ... FROM SNAPSHOT dell'istruzione CREATE FLOW o la clausola incorporata FLOW AUTO CDC di CREATE STREAMING TABLE. L'origine è una clausola obbligatoria FROM SNAPSHOT (snapshot_query) e una clausola opzionale WITH VERSION (version_query) che seleziona la prossima versione snapshot da elaborare. Vedi CREATE FLOW (pipelines).
  • In Python, usa la create_auto_cdc_from_snapshot_flow() funzione per specificare lo snapshot, le chiavi e altri argomenti. Vedi create_auto_cdc_from_snapshot_flow.

Per informazioni dettagliate sui due modelli di inserimento e su quando usarli, vedere Modelli di elaborazione snapshot. Vedere gli esempi di AUTO CDC FROM SNAPSHOT.

Unisci AUTO CDC DA SNAPSHOT e AUTO CDC

La versione snapshot e la colonna di sequenziamento CDC formano un unico dominio di ordinazione. Un evento CDC più recente dell'ultima istantanea ha la precedenza sull'istantanea. Uno snapshot più recente ha la precedenza sugli eventi CDC più vecchi e considera le chiavi assenti dallo snapshot come eliminate nella versione dello snapshot. Gli eventi CDC più recenti possono preservare o ripristinare queste chiavi.

Un obiettivo unificato deve soddisfare i seguenti requisiti:

  • La destinazione utilizza SCD Tipo 1.
  • Il bersaglio ha esattamente un AUTO CDC FROM SNAPSHOT flusso e uno o più AUTO CDC flussi.
  • Ogni flusso AUTO CDC ha un nome unico.
  • Tutti i flussi usano lo stesso numero di chiavi nello stesso ordine. I nomi delle chiavi Snapshot-flow vengono confrontati senza distinzione tra maiuscole e minuscole con i nomi delle chiavi AUTO CDC Flussi multipli AUTO CDC devono utilizzare nomi chiave e casing identici.
  • La versione snapshot e ogni colonna di sequenziamento CDC hanno esattamente lo stesso tipo di dato.
  • Il AUTO CDC FROM SNAPSHOT flusso non definisce le aspettative.
  • I AUTO CDC flussi non utilizzano IGNORE NULL UPDATES o le relative varianti elenco colonne.
  • Il pipeline utilizza la modalità attivata da trigger.
  • Se aggiungi un AUTO CDC flusso a un target esistente AUTO CDC FROM SNAPSHOT , il target non deve contenere colonne utente i cui nomi sono in conflitto con le colonne di sistema riservate AUTO CDC .

Sia i flussi AUTO CDC che AUTO CDC FROM SNAPSHOT possono utilizzare l'interfaccia della pipeline Python o SQL. Puoi mescolare flussi SQL e Python nello stesso target.

Usa once=True in Python o ONCE in SQL sul AUTO CDC FROM SNAPSHOT flusso per popolare retroattivamente uno snapshot una volta mentre i AUTO CDC flussi continuano a elaborare nuovi eventi negli aggiornamenti successivi. Un aggiornamento completo del target riesegue il flusso di snapshot una tantum. Per un esempio completo, vedi Aggiungi un backfill a una tabella AUTO CDC SCD Tipo 1.

Questo pattern non supporta destinazioni SCD di tipo 2 o bitemporali.

Usare più colonne per la sequenziazione

Per sequenziare in base a più colonne, ad esempio un timestamp e un ID per eliminare parità, usare un operatore STRUCT per combinarli. L'API ordina per primo il primo campo e, in caso di parità, considera il secondo campo e così via.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Pitone

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

Esempi di AUTO CDC

Gli esempi seguenti illustrano l'elaborazione SCD tipo 1 e tipo 2 usando una sorgente di flusso di dati delle modifiche. I dati di esempio creano nuovi record utente, eliminano un record utente e aggiornano i record utente. Nell'esempio SCD Type 1 le ultime UPDATE operazioni arrivano in ritardo e vengono eliminate dalla tabella di destinazione, dimostrando la gestione degli eventi fuori ordine.

Di seguito sono riportati i record di input usati in questi esempi. Questi dati vengono creati eseguendo la query nella sezione Creare dati di esempio .

userId nome city operazione numeroSequenza
124 Raul Oaxaca INSERT 1
123 Isabel Monterrey INSERT 1
125 Mercedes Tijuana INSERT 2
126 Lily Cancun INSERT 2
123 nullo nullo DELETE 6
125 Mercedes Guadalajara UPDATE 6
125 Mercedes Mexicali UPDATE 5
123 Isabel Chihuahua UPDATE 5

Quando si rimuove il commento dalla riga finale della query di generazione dei dati di esempio, inserisce il seguente record che specifica di troncare la tabella a sequenceNum=3:

userId nome city operazione numeroSequenza
nullo nullo nullo TRONCARE 3

Annotazioni

Tutti gli esempi seguenti includono opzioni per specificare le operazioni DELETE e TRUNCATE, ma ognuna è facoltativa.

Creare dati di esempio

Eseguire le istruzioni seguenti per creare un set di dati di esempio. Questo codice non deve essere eseguito come parte di una definizione della pipeline. Eseguirlo dalla cartella di esplorazione della pipeline, anziché dalla cartella delle trasformazioni.

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)
);

Elaborare gli aggiornamenti di SCD Tipo 1

ScD Type 1 mantiene solo la versione più recente di ogni record. L'esempio seguente legge dal feed di dati delle modifiche creato in precedenza e applica le modifiche a una destinazione della tabella di streaming. Hai bisogno di una pipeline per eseguire questo codice. Vedere Sviluppare ed eseguire il debug di pipeline ETL con l'editor delle pipeline di Lakeflow.

Pitone

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;

Dopo aver eseguito l'esempio scD Type 1, la tabella di destinazione contiene i record seguenti:

userId nome city
124 Raul Oaxaca
125 Mercedes Guadalajara
126 Lily Cancun

L'utente 123 (Isabel) è stato eliminato e non viene visualizzato. L'utente 125 (Mercedes) mostra solo l'ultima città (Guadalajara) perché SCD Type 1 sovrascrive i valori precedenti. Il precedente UPDATE a sequenceNum=5 è stato eliminato perché è arrivato un successivo aggiornamento a sequenceNum=6.

Dopo aver eseguito l'esempio con il TRUNCATE record senza commento, la tabella viene cancellata in sequenceNum=3. Ciò significa che i record 124 e 126 non si trovano nella tabella e la tabella di destinazione finale contiene solo il record seguente:

userId nome city
125 Mercedes Guadalajara

Elaborare gli aggiornamenti di tipo 2 di SCD

Il tipo SCD 2 mantiene una cronologia completa delle modifiche creando nuove righe per ogni versione di un record, con __START_AT colonne e __END_AT che indicano quando ogni versione era attiva.

Pitone

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;

Dopo aver eseguito l'esempio scD Type 2, la tabella di destinazione contiene i record seguenti:

userId nome city __START_AT __END_AT
123 Isabel Monterrey 1 5
123 Isabel Chihuahua 5 6
124 Raul Oaxaca 1 nullo
125 Mercedes Tijuana 2 5
125 Mercedes Mexicali 5 6
125 Mercedes Guadalajara 6 nullo
126 Lily Cancun 2 nullo

La tabella mantiene la cronologia completa. L'utente 123 ha due versioni (terminate alla sequenza 6 quando sono state eliminate). L'utente 125 ha tre versioni che mostrano le modifiche della città. I record con __END_AT = null sono attualmente attivi.

Tenere traccia di un sottoinsieme di colonne con SCD Type 2

Per impostazione predefinita, scD Type 2 crea una nuova versione ogni volta che viene modificato qualsiasi valore di colonna. È possibile specificare un subset di colonne da tenere traccia, in modo che le modifiche apportate ad altre colonne aggiornino la versione corrente sul posto anziché generare un nuovo record di cronologia.

L'esempio seguente esclude la city colonna dal rilevamento della cronologia:

Pitone

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)

Poiché city le modifiche non vengono rilevate, gli aggiornamenti delle città sovrascrivono la riga corrente anziché creare una nuova versione. La tabella di destinazione contiene i record seguenti:

userId nome city __START_AT __END_AT
123 Isabel Chihuahua 1 6
124 Raul Oaxaca 1 nullo
125 Mercedes Guadalajara 2 nullo
126 Lily Cancun 2 nullo

Esempi di AUTO CDC FROM SNAPSHOT

Nelle sezioni seguenti vengono forniti esempi dell'utilizzo di AUTO CDC FROM SNAPSHOT per processare gli snapshot nelle tabelle di destinazione SCD di tipo 1 o di tipo 2. Per informazioni generali su quando usare questa API, vedere Change Data Capture and snapshots (Modificare acquisizione dati e snapshot).

Gli esempi in questa sezione includono implementazioni di Python e SQL. Per la sintassi SQL completa, vedi CREATE FLOW (pipeline).

Esempio: Processare snapshot periodici

Usa questo approccio quando le istantanee arrivano regolarmente e in ordine. L'interfaccia Python utilizza l'ordine di aggiornamento della pipeline per il versioning. Per SQL, l'esempio utilizza una tabella di versione separata per identificare quando è disponibile un nuovo snapshot.

È possibile leggere gli snapshot da più tipi di origine, tra cui tabelle Delta, file di archiviazione cloud e connessioni JDBC.

Passaggio 1: Creare dati di esempio

Creare una tabella contenente i dati dello snapshot. Esegui il codice seguente da un notebook o da Databricks SQL nella cartella explorations della pipeline.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

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

CREATE TABLE main.cdc_tutorial.snapshot_version (
  version BIGINT
);

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

INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);

Passaggio 2: Eseguire AUTO CDC dallo SNAPSHOT

Hai bisogno di una pipeline per eseguire il codice in questo passaggio. Vedere Sviluppare ed eseguire il debug di pipeline ETL con l'editor delle pipeline di Lakeflow.

Per Python, scegli un tipo sorgente per la vista snapshot (il codice di creazione di esempio genera una tabella Delta):

Opzione A: Lettura da una tabella Delta 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")
Opzione B: Leggere dall'archiviazione cloud 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")
Opzione C: Lettura da JDBC (solo calcolo classico) 
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()
  )

Poi aggiungi la tabella di destinazione e il flusso. L'implementazione SQL utilizza la snapshot_version tabella per identificare quando è disponibile una nuova snapshot:

Pitone

dp.create_streaming_table("target")

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

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
  SELECT version
  FROM main.cdc_tutorial.snapshot_version
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
)
KEYS (userId)
STORED AS SCD TYPE 2;

Dopo la prima esecuzione della pipeline, tutti i record vengono inseriti come righe attive:

userId city __START_AT __END_AT
1 Oaxaca 0 nullo
2 Monterrey 0 nullo
3 Tijuana 0 nullo

Annotazioni

Per usare invece SCD Tipo 1 e mantenere solo lo stato attuale, imposta stored_as_scd_type=1 in Python o STORED AS SCD TYPE 1 SQL. In questo caso, la tabella di destinazione non include le colonne __START_AT e __END_AT.

Passaggio 3: Simulare un nuovo snapshot ed eseguire di nuovo

Aggiorna la tabella sorgente per simulare l'arrivo di una nuova snapshot (esegui questo codice da un notebook o file SQL nella explorations cartella della tua 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');

UPDATE main.cdc_tutorial.snapshot_version SET version = 1;

Esegui di nuovo la pipeline. AUTO CDC FROM SNAPSHOT confronta il nuovo snapshot con quello precedente e rileva che l'utente 1 è stato eliminato, gli utenti 2 e 3 sono stati aggiornati e gli utenti 4 e 6 sono stati inseriti. Viene generato un feed di modifiche e AUTO CDC viene usato per creare la tabella di output.

Dopo la seconda esecuzione con SCD Type 2, la tabella di destinazione contiene i record seguenti:

userId city __START_AT __END_AT
1 Oaxaca 0 1
2 Monterrey 0 1
2 Carmel 1 nullo
3 Tijuana 0 1
3 Los Angeles 1 nullo
4 Valle della Morte 1 nullo
6 Kings Canyon 1 nullo

L'utente 1 è stato terminato (eliminato). Gli utenti 2 e 3 hanno ognuna due versioni che mostrano le modifiche della città. Gli utenti 4 e 6 sono stati appena inseriti.

Dopo la seconda esecuzione con scD Type 1, la tabella di destinazione mostra solo lo stato corrente:

userId city
2 Carmel
3 Los Angeles
4 Valle della Morte
6 Kings Canyon

Esempio: Elaborare snapshot usando le funzioni di versione

Usare questo approccio quando è necessario un controllo esplicito sull'ordinamento degli snapshot. Ad esempio, usare questo approccio quando arrivano più snapshot contemporaneamente o gli snapshot non arrivano in ordine. Definisci come selezionare lo snapshot successivo e il suo numero di versione. L'API elabora gli snapshot in ordine di versione crescente:

  • Se sono in archivio più snapshot, il selettore di esempio elabora gli snapshot disponibili in ordine di versione.
  • Se un’istantanea arriva fuori ordine (ad esempio, snapshot_3 arriva dopo snapshot_4), il selettore di esempi la esclude perché la sua versione non è maggiore dell'ultima versione confermata.
  • Se non ci sono nuovi snapshot, la funzione version o la query non restituisce alcun risultato e non avviene alcuna elaborazione.

Passaggio 1: Preparare i file di snapshot

Creare dei file CSV contenenti dati snapshot e aggiungerli a un volume o a una posizione di archiviazione nel cloud. Denominare i file in ordine cronologico (ad esempio, snapshot_1.csv, snapshot_2.csv).

Ogni file deve contenere colonne per userId e city. Per esempio:

snapshot_1.csv:

userId city
1 Oaxaca
2 Monterrey
3 Tijuana

snapshot_2.csv:

userId city
2 Carmel
3 Los Angeles
4 Valle della Morte

Passaggio 2: Eseguire AUTO CDC FROM SNAPSHOT con una funzione di versione

In una pipeline, crea un nuovo file Python o SQL nella transformations cartella e aggiungi il seguente codice. Quindi esegui la pipeline. Vedere Sviluppare ed eseguire il debug di pipeline ETL con l'editor delle pipeline di Lakeflow.

Pitone

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
)

SQL

CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;

CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'snapshot_',
    CAST((SELECT version FROM current_snapshot_version()) AS STRING),
    '.csv'
  )
)
WITH VERSION (
  WITH snapshot_files AS (
    SELECT
      CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
    WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

Annotazioni

Per usare invece SCD Tipo 1, imposta stored_as_scd_type=1 in Python o STORED AS SCD TYPE 1 SQL.

Dopo l'elaborazione snapshot_1.csv, la tabella di destinazione contiene i record seguenti:

userId city __START_AT __END_AT
1 Oaxaca 1 nullo
2 Monterrey 1 nullo
3 Tijuana 1 nullo

Dopo l'elaborazione snapshot_2.csv, la tabella di destinazione contiene i record seguenti:

userId city __START_AT __END_AT
1 Oaxaca 1 2
2 Monterrey 1 2
2 Carmel 2 nullo
3 Tijuana 1 2
3 Los Angeles 2 nullo
4 Valle della Morte 2 nullo

Annotazioni

Tenere presente che, per scD Type 1, la tabella è esattamente simile allo snapshot più recente. La differenza è che le query downstream possono usare il feed di modifiche per elaborare solo i record modificati.

Passaggio 3: Aggiungere nuovi snapshot

Aggiungere un nuovo file CSV al percorso di archiviazione con dati modificati( ad esempio, valori città modificati, nuove righe o righe rimosse). Eseguire quindi di nuovo la pipeline per elaborare il nuovo snapshot.

Limitazioni

Risorse aggiuntive

  • Ultimo aggiornamento: