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

Deklarativní potrubí Lakeflow Spark (SDP) zjednodušují zachycování změn dat (CDC) pomocí AUTO CDC a AUTO CDC FROM SNAPSHOT API. 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ím sledování naleznete v tématu Jak funguje bitemporální 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 jak funguje bitemporal AUTO CDC.

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ý protokol SDP nebo SDPPro nebo Advancededice.

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.

Jak funguje bitemporal AUTO CDC

Important

Bitemporal AUTO CDC je v beta verzi.

SCD Typ 1 a Typ 2 jsou jednotná: sledují změny v rámci jedné časové dimenze. Bitemporal rozšiřuje historii typu 2 SCD, aby sledovala změny ve dvou časových dimenzích a rozlišovala mezi dvěma perspektivami:

  • Pracovní doba: kdy k události skutečně došlo.
  • Systémový čas: kdy systém zaznamenal nebo ingestuje událost.

Podobně jako SCD Type 2 bitemporal zachovává úplnou historii záznamů. Přidá druhou časovou osu, abyste mohli rekonstruovat jak data, tak i to, co systém v jakémkoli okamžiku v minulosti věřil.

Například investiční fond ingestuje data akcií ze zdrojového systému. Cena akcií společnosti Acme Corp se změní 1. ledna, ale fond tuto aktualizaci zpracuje až 5. ledna. Bitemporal AUTO CDC umožňuje fondu zodpovědět dvě odlišné otázky: jaká skutečná cena akcií Acme Corp byla 1. ledna (pracovní čas) a jakou cenu systém věřil, když fond učinil rozhodnutí o obchodování 3. ledna (systémový čas). Schopnost rozlišovat mezi těmito časovými hledisky je užitečná pro audit, regulatorní výkaznictví a přijímání finančních rozhodnutí.

Pokud chcete povolit bitemporální zpracování, nastavte STORED AS BITEMPORAL (SQL) nebo stored_as_scd_type="bitemporal" (Python), použijte SEQUENCE BY sloupec obchodního času a použijte SYSTEM SEQUENCE BY pro sloupec systémový čas. Cílová tabulka přidává sloupce __SYSTEM_START_AT a __SYSTEM_END_AT vedle sloupců SCD typu 2 __START_AT a __END_AT. Pro podrobnosti o syntaxi se podívejte na AUTO CDC INTO (pipelines) nebo create_auto_cdc_flow.

Podrobný návod, jak vložení, aktualizace, aktualizace mimo pořadí a odstranění ovlivňují bitemporální tabulku, najdete v příkladech bitemporal AUTO CDC.

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.

Příklady pro Bitemporal AUTO CDC

Important

Bitemporal AUTO CDC je v beta verzi.

Následující příklad vytváří bitemporální cílovou tabulku z menší sady syntetických událostí CDC. Sloupec bt má pracovní čas a st sloupec má systémový čas.

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;

Následující kroky ukazují, jak bitemporální tabulka zaznamenává vložení, aktualizace, aktualizace mimo pořadí a odstranění pro jednu společnost. Sloupec sekvencování generuje __START_AT sloupce a __END_AT (pracovní čas) a sloupec pořadí systému generuje __SYSTEM_START_AT sloupce a __SYSTEM_END_AT (systémový čas):

Column Description
__START_AT Pracovní doba, ve které se tento řádek stal platným.
__END_AT Pracovní doba, ve které platnost tohoto řádku končí. null pokud je platnost neurčitá.
__SYSTEM_START_AT Systémový čas, ve kterém je známo, že data tohoto řádku a interval obchodního času jsou platné.
__SYSTEM_END_AT Systémový čas, v němž je známo, že data tohoto řádku a interval obchodního času jsou zneplatněny. null pokud je známo, že je pravda neomezeně dlouho.

Systém zpracovává události, které přicházejí v libovolném pořadí na obou časových osách. Když událost dorazí s dřívějším časem platnosti nebo systémovým časem než u již zpracovaných událostí, systém opraví dotčenou historii, místo aby ji pouze připojil na konec.

Krok 1: Vložení

Společnost A je přidána v 18. 7. 2025 10:01:00 (obchodní čas), ale je zpracována až v 10:05:00 (systémový čas).

Vstup:

Id společnosti Datový bod Klasifikace Sekvencování systému Operation
A XFv1 7/18/2025 10:01:00 7/18/2025 10:05:00 INSERT

Výstup:

Id společnosti Datový bod __START_AT __END_AT __SYSTEM_START_AT __SYSTEM_END_AT
A XFv1 7/18/2025 10:01:00 NULL 7/18/2025 10:05:00 NULL

XFv1 je platný od 10:01:00 bez známého konce. Systém se dozvěděl o této skutečnosti v systémovém čase 10:05:00 bez známého konce.

Krok 2: Aktualizace

Společnost A byla aktualizována dne 18. 7. 2025 v 12:15:43 (obchodní čas) a systém tuto událost zpracuje ve 12:20:00 (systémový čas). Systém zachovává jak to, co věřil před aktualizací, tak i opravenou obchodní historii po ingestované aktualizaci.

Vstup:

Id společnosti Datový bod Klasifikace Sekvencování systému Operation
A XFv2 7/18/2025 12:15:43 7/18/2025 12:20:00 UPDATE

Výstup:

Id společnosti Datový bod __START_AT __END_AT __SYSTEM_START_AT __SYSTEM_END_AT
A XFv1 7/18/2025 10:01:00 NULL 7/18/2025 10:05:00 7/18/2025 12:20:00
A XFv1 7/18/2025 10:01:00 7/18/2025 12:15:43 7/18/2025 12:20:00 NULL
A XFv2 7/18/2025 12:15:43 NULL 7/18/2025 12:20:00 NULL

XFv1 byl považován za platný od 10:01:00 bez známého konce, a systém držel toto přesvědčení od 10:05:00 do 12:20:00. O XFv1 je nyní známo, že je platná pouze do 12:15:43; opravená historie je platná od systémového času 12:20:00 bez známého času ukončení. XFv2 je platný od 12:15:43 bez známého konce a byl naučen v systémovém čase 12:20:00.

Krok 3: Aktualizace mimo objednávku

Dorazí aktualizace mimo pořadí, která uvádí, že společnost A byla ve skutečnosti aktualizována 18. 7. 2025 v 12:05:00 (obchodní čas), ale do systému je načtena až v 12:25:00 (systémový čas). Když aktualizace dorazí později v systémovém čase, ale s dřívějším obchodním časem, systém opraví historický obchodní čas a zachová jak to, co před aktualizací doručenou mimo pořadí považoval za platné, tak i opravenou historii.

Vstup:

Id společnosti Datový bod Klasifikace Sekvencování systému Operation
A XFv3 7/18/2025 12:05:00 7/18/2025 12:25:00 UPDATE

Výstup:

Id společnosti Datový bod __START_AT __END_AT __SYSTEM_START_AT __SYSTEM_END_AT
A XFv1 7/18/2025 10:01:00 NULL 7/18/2025 10:05:00 7/18/2025 12:20:00
A XFv1 7/18/2025 10:01:00 7/18/2025 12:15:43 7/18/2025 12:20:00 7/18/2025 12:25:00
A XFv1 7/18/2025 10:01:00 7/18/2025 12:05:00 7/18/2025 12:25:00 NULL
A XFv3 7/18/2025 12:05:00 7/18/2025 12:15:43 7/18/2025 12:25:00 NULL
A XFv2 7/18/2025 12:15:43 NULL 7/18/2025 12:20:00 NULL

XFv1 bylo považováno za platné od 10:01:00 do 12:15:43 a tento předpoklad je nyní v systémovém čase platný do 12:25:00. Nová aktualizace opravuje obchodní platnost XFv1 tak, aby končila v 12:05:00, a opravuje historii s účinností od systémového času 12:25:00. O XFv3 je nyní známo, že platilo od 12:05:00 do 12:15:43, přičemž tato znalost je v systémovém čase platná od 12:25:00 bez známého konce platnosti.

Krok 4: Odstranění

Společnost A je odstraněna 18. 7. 2025 v 12:30:00 a systém tuto událost zpracuje v 12:30:00. Vzhledem k tomu, že operace odstranění představuje konec obchodní existence entity, systém vytvoří žádný náhradní řádek. XFv2 se zobrazuje ve dvou řádcích, přičemž je zachována úplná auditní stopa jak okamžiku, kdy společnost zanikla, tak okamžiku, kdy se systém o odstranění dozvěděl.

Vstup:

Id společnosti Datový bod Klasifikace Sekvencování systému Operation
A XFv2 7/18/2025 12:30:00 7/18/2025 12:30:00 DELETE

Výstup:

Id společnosti Datový bod __START_AT __END_AT __SYSTEM_START_AT __SYSTEM_END_AT
A XFv1 7/18/2025 10:01:00 NULL 7/18/2025 10:05:00 7/18/2025 12:20:00
A XFv1 7/18/2025 10:01:00 7/18/2025 12:15:43 7/18/2025 12:20:00 7/18/2025 12:25:00
A XFv1 7/18/2025 10:01:00 7/18/2025 12:05:00 7/18/2025 12:25:00 NULL
A XFv3 7/18/2025 12:05:00 7/18/2025 12:15:43 7/18/2025 12:25:00 NULL
A XFv2 7/18/2025 12:15:43 NULL 7/18/2025 12:20:00 7/18/2025 12:30:00
A XFv2 7/18/2025 12:15:43 7/18/2025 12:30:00 7/18/2025 12:30:00 NULL

XFv2 byl platný od 12:15:43 bez známého času ukončení a systém tento stav považoval za platný od 12:20:00 do 12:30:00. Po zpracování odstranění je o XFv2 známo, že je platné pouze do 12:30:00; jedná se o opravenou historii s účinností od systémového času 12:30:00.

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