Rozhraní API AUTO CDC: Zjednodušení zachytávání dat změn pomocí datových toků

Kanály Lakeflow zjednodušují zachytávání změn dat (CDC) pomocí rozhraní API AUTO CDC a AUTO CDC FROM SNAPSHOT. Tato rozhraní API automatizují složitost výpočtů pomalu se měnících dimenzí typu 1 a typu 2 ze zdroje CDC nebo snímků databáze. Rozhraní AUTO CDC API také podporuje bitemporální sledování, které zaznamenává změny ve dvou časových dimenzích (Beta). Další informace o SCD typu 1 a typu 2 najdete v tématu Zachytávání změn dat a snímky. Další informace o bitemporální sledování naleznete v tématu Bitemporal AUTO CDC.

Poznámka:

Rozhraní AUTO CDC API nahrazují APPLY CHANGES rozhraní API a mají stejnou syntaxi. Rozhraní APPLY CHANGES API jsou stále dostupná, ale Databricks doporučuje namísto nich používat rozhraní AUTO CDC API.

Rozhraní API, které používáte, závisí na zdroji dat změn:

  • AUTO CDC: Tuto možnost použijte, pokud má zdrojová databáze povolený informační kanál CDC. AUTO CDC zpracovává změny z datového kanálu změn (CDF). Podporuje se v rozhraních SQL pipeline i Pythonu.
  • AUTO CDC FROM SNAPSHOT: Použijte to, když není ve zdrojové databázi povolené CDC a jsou k dispozici pouze snímky. Toto rozhraní API porovnává snímky a zjišťuje změny a pak je zpracovává. Podporuje se v rozhraních SQL pipeline i Pythonu.

U cílů typu SCD 1 můžete také kombinovat AUTO CDC FROM SNAPSHOT tok s jedním nebo více AUTO CDC toky. Tento vzor použijte, když zdroj poskytuje periodické autoritativní snímky a CDC kanál s nižší latencí mezi snímky.

Obě rozhraní API podporují aktualizaci tabulek pomocí SCD Type 1 a Type 2:

  • K přímé aktualizaci záznamů použijte SCD Type 1. Historie se neuchovává pro aktualizované záznamy.
  • Pomocí příkazu SCD Type 2 zachovejte historii záznamů, a to buď u všech aktualizací, nebo aktualizací v zadané sadě sloupců.

Pouze pro AUTO CDC můžete také použít bitemporální úložiště, které rozšiřuje historii SCD Type 2 o sledování změn ve dvou časových dimenzích: platném čase a systémovém čase. Bitemporal je v beta verzi. Viz Bitemporal AUTO CDC.

AUTO CDC podporuje také částečné aktualizace, kdy záznam změn aktualizuje pouze podmnožinu sloupců. Viz Použití částečných aktualizací.

Deklarativní AUTO CDC kanály Apache Sparku nepodporují rozhraní API.

Syntaxi a další informace najdete v tématu AUTO CDC INTO (datové toky), create_auto_cdc_flow, a create_auto_cdc_from_snapshot_flow.

Poznámka:

Použijte tato API k aktualizaci tabulek ve vašich pipelinech na základě změn ve zdrojových datech. Informace o záznamu a dotazování informací o změnách na úrovni řádků pro tabulky Delta najdete v tématu Použití datového kanálu změn v Azure Databricks.

Požadavky

Pokud chcete používat rozhraní API CDC, musí být váš kanál nakonfigurovaný tak, aby používal bezserverové kanály Lakeflow nebo kanály Pro nebo Advancededice Lakeflow.

Jak funguje AUTO CDC

Pokud chcete provést zpracování CDC pomocí AUTO CDC, vytvořte streamovací tabulku a pak pomocí AUTO CDC ... INTO příkazu v SQL nebo create_auto_cdc_flow() funkci v Pythonu určete zdroj, klíče a sekvencování kanálu změn. Vysvětlení toho, jak funguje sekvencování a logika SCD, najdete v tématu Změna zachytávání a snímků dat. Podívejte se na příklady AUTO CDC.

Pro počáteční hydraci ze zdroje s kanálem změn použijte AUTO CDConce tok a pokračujte ve zpracování kanálu změn. Viz Replikace externí tabulky RDBMS pomocí auto CDC.

Pro podrobnosti o syntaxi se podívejte na AUTO CDC INTO (pipelines) nebo create_auto_cdc_flow.

Jak funguje AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT určuje změny ve zdrojových datech porovnáním snímků v pořadí. Snímky můžete číst přímo z tabulky Delta, souborů cloudového úložiště nebo JDBC. Podporuje se v rozhraních SQL pipeline i Pythonu.

Pro provedení CDC zpracování s AUTO CDC FROM SNAPSHOT, vytvořte streamovací tabulku a poté definujte tok:

  • V SQL použijte AUTO CDC ... FROM SNAPSHOT tvar příkazu CREATE FLOW nebo vloženou klauzuli FLOW AUTO CDC příkazu CREATE STREAMING TABLE. Zdroj tvoří povinná FROM SNAPSHOT (snapshot_query) klauzule a volitelná WITH VERSION (version_query) klauzule, která vybírá další verzi snímku ke zpracování. Viz CREATE FLOW (pipelines).
  • V Python použijte create_auto_cdc_from_snapshot_flow() funkci k určení snapshotu, klíčů a dalších argumentů. Viz create_auto_cdc_from_snapshot_flow.

Podrobnosti o těchto dvou vzorech příjmu dat a o tom, kdy je použít, najdete v tématu Vzory zpracování snímků. Podívejte se na příklady AUTO CDC FROM SNAPSHOT.

Spojte AUTO CDC ZE SNAPSHOT a AUTO CDC

Snapshot verze a sloupec sekvenování CDC tvoří jednu pořadací doménu. Událost CDC novější než nejnovější snapshot má přednost před snapshotem. Novější snapshot má přednost před staršími událostmi CDC a klíče, které v snapshotu chybí, považuje za smazané ve verzi snapshotu. Novější akce CDC mohou tyto klíče uchovávat nebo obnovit.

Jednotný cíl musí splňovat následující požadavky:

  • Cílový systém používá SCD typ 1.
  • Cílový objekt má přesně jeden AUTO CDC FROM SNAPSHOT tok a jeden nebo více AUTO CDC toků.
  • Každý AUTO CDC tok má svůj jedinečný název.
  • Všechny toky používají stejný počet klíčů ve stejném pořadí. Názvy klíčů typu snapshot-flow se porovnávají bez rozlišení velkých a malých písmen s názvy klíčů AUTO CDC. Více AUTO CDC toků musí používat stejná jména klíčů a zápis velkých a malých písmen.
  • Verze snapshot a každý sloupec sekvenování CDC mají přesně stejný datový typ.
  • tok AUTO CDC FROM SNAPSHOT neurčuje očekávání.
  • Postupy AUTO CDC nepoužívají IGNORE NULL UPDATES ani jeho varianty typu seznamu sloupců.
  • Pipeline používá spouštěcí režim.
  • Pokud přidáte AUTO CDC tok do existujícího AUTO CDC FROM SNAPSHOT cíle, cíl nesmí obsahovat uživatelské sloupce, jejichž názvy kolidují s názvy AUTO CDC vyhrazených systémových sloupců.

Oba toky AUTO CDC i AUTO CDC FROM SNAPSHOT mohou využívat rozhraní pipeline Python nebo SQL. Můžete kombinovat toky SQL a Python ve stejném cíli.

Použijte once=True v Pythonu nebo ONCE v SQL v toku AUTO CDC FROM SNAPSHOT k doplnění snapshotu jednou, zatímco AUTO CDC flow pokračuje ve zpracování nových událostí při dalších aktualizacích. Plné obnovení cíle znovu spustí proces jednorázového snímku. Úplný příklad najdete v části Přidání zpětného doplnění do tabulky AUTO CDC SCD typu 1.

Tento vzor nepodporuje SCD typ 2 ani bitemporální cíle.

Použití více sloupců pro sekvencování

Pokud chcete sekvencovat podle více sloupců (například časové razítko a ID pro přerušení vazeb), použijte STRUCT je ke kombinování. Rozhraní API řadí nejprve podle prvního pole a v případě shody zohledňuje druhé pole, a tak dále.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

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

Příklady automatického CDC

Následující příklady ukazují zpracování scd typu 1 a typu 2 pomocí zdroje dat změn. Ukázková data vytvoří nové záznamy uživatelů, odstraní záznam uživatele a aktualizuje záznamy uživatelů. V příkladu typu SCD 1 se poslední UPDATE operace zpozdí a jsou vyřazeny z cílové tabulky, což demonstruje zpracování událostí v nesprávném pořadí.

Tady jsou vstupní záznamy použité v těchto příkladech. Tato data se vytvoří spuštěním dotazu v části Vytvořit ukázková data .

userId název město operace číslo sekvence
124 Raul Oaxaca INSERT 1
123 Isabel Monterrey INSERT 1
125 Mercedes Tijuana INSERT 2
126 Lilie Cancun INSERT 2
123 null null DELETE 6
125 Mercedes Guadalajara UPDATE 6
125 Mercedes Mexicali UPDATE 5
123 Isabel Chihuahua UPDATE 5

Pokud odkomentujete poslední řádek v dotazu na generování ukázkových dat, vloží následující záznam, který určuje, že se má tabulka zkrátit (vymazat tabulku) na sequenceNum=3adrese:

userId název město operace číslo sekvence
null null null ZKRÁTIT 3

Poznámka:

Všechny následující příklady zahrnují možnosti pro určení obou DELETETRUNCATE operací, ale každý z nich je volitelný.

Vytvoření ukázkových dat

Spuštěním následujících příkazů vytvořte ukázkovou datovou sadu. Tento kód není určen ke spuštění jako součást definice kanálu. Spusťte ho ze složky průzkumu datového toku, a ne z adresáře transformací.

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

Zpracování aktualizací SCD typu 1

ScD Type 1 uchovává pouze nejnovější verzi každého záznamu. Následující příklad načte z datového kanálu změn vytvořeného výše a aplikuje změny na cílovou streamovací tabulku. Potřebujete pipeline, abyste mohli tento kód spustit. Viz Vývoj a ladění kanálů ETL pomocí Editoru kanálů Lakeflow.

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;

Po spuštění příkladu typu 1 typu SCD obsahuje cílová tabulka následující záznamy:

userId název město
124 Raul Oaxaca
125 Mercedes Guadalajara
126 Lilie Cancun

Uživatel 123 (Isabel) byl odstraněn a nezobrazí se. Uživatel 125 (Mercedes) zobrazuje pouze nejnovější město (Guadalajara), protože SCD Type 1 přepíše předchozí hodnoty. Předchozí verze na UPDATEsequenceNum=5 byla zrušena, protože přišla novější aktualizace na sequenceNum=6.

Po spuštění příkladu s nekommentovaným záznamem TRUNCATE se tabulka vymaže na sequenceNum=3adrese . To znamená, že záznamy 124 a 126 nejsou v tabulce a konečná cílová tabulka obsahuje pouze následující záznam:

userId název město
125 Mercedes Guadalajara

Zpracování aktualizací SCD typu 2

ScD Type 2 zachovává úplnou historii změn vytvořením nových řádků pro každou verzi záznamu a __START_AT__END_AT sloupci označujícími, kdy byla každá verze aktivní.

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;

Po spuštění příkladu typu SCD 2 obsahuje cílová tabulka následující záznamy:

userId název město __START_AT __END_AT
123 Isabel Monterrey 1 5
123 Isabel Chihuahua 5 6
124 Raul Oaxaca 1 null
125 Mercedes Tijuana 2 5
125 Mercedes Mexicali 5 6
125 Mercedes Guadalajara 6 null
126 Lilie Cancun 2 null

Tabulka zachovává celou historii. Uživatel 123 má dvě verze (před odstraněním skončila v pořadí 6). Uživatel 125 má tři verze zobrazující změny města. Záznamy s __END_AT = null jsou aktuálně aktivní.

Sledování podmnožiny sloupců pomocí SCD Type 2

Ve výchozím nastavení vytvoří SCD Type 2 novou verzi pokaždé, když se změní jakákoli hodnota sloupce. Můžete zadat podmnožinu sloupců, které se mají sledovat, aby změny v jiných sloupcích aktualizovaly aktuální verzi místo generování nového záznamu historie.

Následující příklad vyloučí city sloupec ze zaznamenávání historie.

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)

Vzhledem k tomu city , že změny nejsou sledovány, aktualizace měst přepíší aktuální řádek místo vytvoření nové verze. Cílová tabulka obsahuje následující záznamy:

userId název město __START_AT __END_AT
123 Isabel Chihuahua 1 6
124 Raul Oaxaca 1 null
125 Mercedes Guadalajara 2 null
126 Lilie Cancun 2 null

Příklady AUTO CDC FROM SNAPSHOT

Následující části obsahují příklady použití AUTO CDC FROM SNAPSHOT ke zpracování snímků do cílových tabulek SCD Type 1 nebo Type 2. Informace o tom, kdy použít toto rozhraní API, najdete v tématu Změna zachytávání a snímků dat.

Příklady v této sekci zahrnují implementace Python a SQL. Pro kompletní SQL syntaxi viz CREATE FLOW (pipelines).

Příklad: Zpracování periodických snímků

Tento přístup používejte, když snímky přicházejí pravidelně a ve správném pořadí. Rozhraní Python používá pořadí aktualizací pipeline pro verzování. Pro SQL příklad používá samostatnou tabulku verzí k identifikaci, kdy je k dispozici nový snapshot.

Snímky můžete číst z více typů zdrojů, včetně tabulek Delta, souborů cloudového úložiště a připojení JDBC.

Krok 1: Vytvoření ukázkových dat

Vytvořte tabulku obsahující data snímku. Ve složce pipeline spusťte následující kód z poznámkového bloku nebo z Databricks SQL explorations.

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

Krok 2: Spuštění AUTOMATICKÉho CDC ZE SNÍMKU

V tomto kroku potřebujete pipeline, abyste mohli kód spustit. Viz Vývoj a ladění kanálů ETL pomocí Editoru kanálů Lakeflow.

Pro Python zvolte zdrojový typ pro zobrazení snapshotu (ukázkový kód pro vytvoření vygeneruje tabulku Delta):

Možnost A: Čtení z tabulky Delta 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")
Možnost B: Čtení z cloudového úložiště 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")
Možnost C: Čtení z JDBC (pouze klasické výpočetní prostředky) 
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()
  )

Pak přidejte cílovou tabulku a postup. Implementace SQL používá snapshot_version tabulku k identifikaci, kdy je k dispozici nový snapshot:

Python

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;

Po prvním spuštění kanálu se všechny záznamy vloží jako aktivní řádky:

userId město __START_AT __END_AT
1 Oaxaca 0 null
2 Monterrey 0 null
3 Tijuana 0 null

Poznámka:

Chcete-li místo toho použít SCD Type 1 a ponechat pouze aktuální stav, nastavte stored_as_scd_type=1 v Pythonu nebo STORED AS SCD TYPE 1 v SQL. V tomto případě cílová tabulka neobsahuje __START_AT a __END_AT sloupce.

Krok 3: Simulace nového snímku a opětovné spuštění

Aktualizujte zdrojovou tabulku, aby simulovala příchod nového snapshotu (spusťte tento kód z notebooku nebo SQL souboru ve složce explorations vašeho 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;

Opětovné spuštění kanálu AUTO CDC FROM SNAPSHOT porovná nový snímek s předchozím snímkem a zjistí, že uživatel 1 byl odstraněn, uživatelé 2 a 3 se aktualizovali a uživatelé 4 a 6 se vložili. Tím se vygeneruje kanál změn a AUTO CDC se použije k vytvoření výstupní tabulky.

Po druhém spuštění při použití SCD Type 2 obsahuje cílová tabulka následující záznamy:

userId město __START_AT __END_AT
1 Oaxaca 0 1
2 Monterrey 0 1
2 Carmel 1 null
3 Tijuana 0 1
3 Los Angeles 1 null
4 Údolí smrti 1 null
6 Kings Canyon 1 null

Uživatel 1 byl ukončen (odstraněn). Uživatelé 2 a 3 mají dvě verze zobrazující změny města. Uživatelé 4 a 6 byli nově vloženi.

Po druhém spuštění s scd typem 1 se v cílové tabulce zobrazí pouze aktuální stav:

userId město
2 Carmel
3 Los Angeles
4 Údolí smrti
6 Kings Canyon

Příklad: Zpracování snímků pomocí funkcí verzí

Tento přístup použijte, pokud potřebujete explicitní kontrolu nad řazením snímků. Tento přístup můžete použít například v případě, že se najednou dorazí více snímků nebo snímky dorazí mimo pořadí. Určujete, jak vybrat další snímek a jeho číslo verze. Rozhraní API zpracovává snímky ve vzestupném pořadí verzí:

  • Pokud je v úložišti více snapshotů, výběrový modul příkladu zpracovává dostupné snapshoty v pořadí verzí.
  • Pokud snapshot dorazí mimo pořadí (například snapshot_3 dorazí po snapshot_4), vzorový selektor jej vylučuje, protože jeho verze není větší než poslední zavázaná verze.
  • Pokud nejsou žádné nové snímky, funkce version nebo dotaz nevrátí žádný výsledek a žádné zpracování neprobíhá.

Krok 1: Příprava snímků souborů

Vytvořte soubory CSV obsahující data snímku a přidejte je do volumového nebo cloudového úložiště. Pojmenujte soubory chronologicky (například snapshot_1.csv, snapshot_2.csv).

Každý soubor by měl obsahovat sloupce pro userId a city. Například:

snapshot_1.csv:

userId město
1 Oaxaca
2 Monterrey
3 Tijuana

snapshot_2.csv:

userId město
2 Carmel
3 Los Angeles
4 Údolí smrti

Krok 2: Spuštění funkce AUTO CDC FROM SNAPSHOT s verzní funkcí

V pipeline vytvořte nový Python nebo SQL soubor ve složce transformations a přidejte následující kód. Pak kanál spusťte. Viz Vývoj a ladění kanálů ETL pomocí Editoru kanálů Lakeflow.

Python

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;

Poznámka:

Místo toho použijte SCD Type 1, nastavte stored_as_scd_type=1 v Pythonu nebo STORED AS SCD TYPE 1 v SQL.

Po zpracování snapshot_1.csvobsahuje cílová tabulka následující záznamy:

userId město __START_AT __END_AT
1 Oaxaca 1 null
2 Monterrey 1 null
3 Tijuana 1 null

Po zpracování snapshot_2.csvobsahuje cílová tabulka následující záznamy:

userId město __START_AT __END_AT
1 Oaxaca 1 2
2 Monterrey 1 2
2 Carmel 2 null
3 Tijuana 1 2
3 Los Angeles 2 null
4 Údolí smrti 2 null

Poznámka:

Mějte na paměti, že pro SCD Type 1 tabulka vypadá přesně jako poslední snímek. Rozdíl je v tom, že podřízené dotazy můžou kanál změn používat pouze ke zpracování změněných záznamů.

Krok 3: Přidání nových snímků

Přidejte nový soubor CSV do umístění úložiště s upravenými daty (například změněné názvy měst, nové řádky nebo odebrané řádky). Potom znovu spusťte kanál a zpracujte nový snímek.

Omezení

Dodatečné zdroje

  • Poslední aktualizace v