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 pouze v rozhraní Pythonu.

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:

Tato stránka popisuje, jak aktualizovat tabulky v pipelinách 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í. Podporováno pouze v rozhraní pipeline Pythonu. Snímky můžete číst přímo z tabulky Delta, souborů cloudového úložiště nebo JDBC.

Chcete-li provést zpracování CDC pomocí AUTO CDC FROM SNAPSHOT, vytvořte streamovací tabulku a pak pomocí create_auto_cdc_from_snapshot_flow() funkce určete snímek, klíče a další argumenty. 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.

Podrobnosti o syntaxi najdete v tématu create_auto_cdc_from_snapshot_flow.

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. Co jsou kanály? a spusťte tento kód.

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říklad: Zpracování snímků pomocí času příjmu datového kanálu

Tento přístup použijte, když snímky přicházejí pravidelně a v pořadí a při správě verzí lze se spoléhat na časové razítko běhu pipeline. Při každé aktualizaci potrubí se ingestuje nový snímek.

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

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

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

Co jsou kanály? a spusťte kód v tomto kroku.

Zvolte typ zdroje pro zobrazení snímků (vzorový 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()
  )

Všechny možnosti, zapište do cíle

Pak přidejte cílovou tabulku a tok:

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  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 zachovat pouze aktuální stav, nastavte stored_as_scd_type=1. 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 nový snímek přicházející (spusťte tento kód z notebooku nebo SQL souboru ve složce explorations vaší pipeliny):

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

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í. Napíšete funkci, která určuje, který snímek se má zpracovat, a jeho číslo verze. Rozhraní API zpracovává snímky ve vzestupném pořadí verzí:

  • Pokud je v úložišti více snímků, všechny se zpracovávají v pořadí.
  • Pokud snímek dorazí mimo pořadí (například snapshot_3 po doručení snapshot_4), přeskočí se.
  • Pokud nejsou žádné nové snímky, funkce se vrátí None a nedojde k žádnému zpracování.

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í

Vytvořte nový poznámkový blok a vložte následující kód pipeline. Co jsou kanály?

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
)

Poznámka:

Pokud chcete místo toho použít SCD Type 1, nastavte stored_as_scd_type=1.

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í

  • Sloupec sekvencování musí být řaditelný datový typ. NULL Hodnoty sekvencování nejsou podporovány.
  • AUTO CDC FROM SNAPSHOT je podporován pouze v rozhraní kanálu Pythonu; rozhraní SQL není podporováno.
  • Pokud chcete streamovat data z cíle procesu AUTO CDC, přečtěte si z jeho kanálu změn. Podrobnosti najdete v tématu Čtení datového kanálu změn z cílové tabulky AUTO CDC.

Dodatečné zdroje

  • Last updated on