Pokročilá témata AUTO CDC

Kromě základních rozhraní API AUTO CDC a AUTO CDC FROM SNAPSHOT můžete spouštět operace DML nad cílovými tabulkami, číst kanály změnových dat z cílů CDC, sledovat metriky zpracování, aplikovat částečné aktualizace a sledovat změny pomocí bitemporálního úložiště. Úvod k rozhraním AUTO CDC API najdete v AUTO CDC rozhraní API: Zjednodušení zachytávání změn dat pomocí kanálů.

Přidání, změna nebo odstranění dat v cílové streamovací tabulce

Pokud kanál publikuje tabulky do katalogu Unity, můžete použít příkazy jazyka DML ( Data Manipulat Language ), včetně příkazů vložení, aktualizace, odstranění a sloučení, a upravit cílové tabulky streamování vytvořené příkazy AUTO CDC ... INTO .

Poznámka:

  • Příkazy DML, které upravují schéma tabulky streamované tabulky, nejsou podporovány. Ujistěte se, že se příkazy DML nepokoušnou vyvíjet schéma tabulky.
  • Příkazy DML, které aktualizují streamovací tabulku, lze spustit pouze ve sdíleném clusteru Unity Catalog nebo v SQL warehouse s použitím Databricks Runtime 13.3 LTS a vyšší.
  • Vzhledem k tomu, že streamování vyžaduje pouze připojované zdroje dat, vyžaduje-li vaše zpracování streamování ze zdrojové streamovací tabulky, která obsahuje změny (například prostřednictvím příkazů DML), nastavte příznak skipChangeCommits při čtení zdrojové streamovací tabulky. Při nastavení skipChangeCommits budou transakce, které odstraňují nebo upravují záznamy ve zdrojové tabulce, ignorovány. Pokud vaše zpracování nevyžaduje streamovací tabulku, můžete jako cílovou tabulku použít materializované zobrazení (které nemá omezení pouze pro přidání).

Protože kanál používá zadaný SEQUENCE BY sloupec a šíří odpovídající hodnoty sekvencování do __START_AT a __END_AT sloupců cílové tabulky (pro SCD Type 2), je nutné zajistit, aby příkazy DML používaly platné hodnoty pro tyto sloupce, aby se zachovalo správné pořadí záznamů. Podívejte se, jak AUTO CDC funguje.

Další informace o použití příkazů DML se streamovanými tabulkami najdete v tématu Přidání, změna nebo odstranění dat v streamované tabulce.

Následující příklad vloží aktivní záznam s počáteční sekvencí 5:

INSERT INTO my_streaming_table (id, name, __START_AT, __END_AT) VALUES (123, 'John Doe', 5, NULL);

Návod

Pokud potřebujete přejmenovat __START_AT sloupce __END_AT v cílové tabulce SCD Type 2 (například aby odpovídaly požadavkům podřízeného schématu), vytvořte zobrazení nad cílovou tabulkou:

CREATE VIEW my_employees_view AS
SELECT
  *,
  __START_AT AS valid_from,
  __END_AT AS valid_to
FROM my_scd2_target_table;

Přečíst datový vstup změn z cílové tabulky AUTO CDC

Ve službě Databricks Runtime 15.2 a novějších můžete číst datový proud změn z tabulky streamování, na kterou jsou směrovány dotazy AUTO CDC nebo AUTO CDC FROM SNAPSHOT, stejným způsobem jako čtete datový proud změn z jiných tabulek Delta. Pro čtení datového kanálu změn z cílové tabulky streamování jsou potřeba následující:

  • Cílová streamovací tabulka musí být publikovaná v katalogu Unity. Podívejte se na Použití katalogu Unity s kanály.
  • Pokud chcete číst datový kanál změn z cílové streamovací tabulky, musíte použít Databricks Runtime 15.2 nebo vyšší. Pokud chcete číst datový kanál změn v jiném potrubí, musí být potrubí nakonfigurované tak, aby používalo Databricks Runtime 15.2 nebo vyšší.

Zdroj dat o změnách čtete z cílové streamovací tabulky vytvořené v pipeline Lakeflow stejným způsobem jako zdroj dat o změnách z jiných tabulek Delta. Další informace o používání funkce Delta change data feed, včetně příkladů v jazycích Python a SQL, najdete v tématu Použití change data feed ve službě Azure Databricks.

Poznámka:

Záznam datového kanálu změn obsahuje metadata identifikující typ události změny. Když se záznam aktualizuje v tabulce, metadata přidružených záznamů změn obvykle obsahují _change_type hodnoty nastavené na update_preimage a update_postimage události.

Tyto hodnoty se ale liší, pokud jsou provedeny aktualizace cílové tabulky streamování, které zahrnují změnu hodnot primárního klíče _change_type. Pokud změny zahrnují aktualizace primárních klíčů, pole metadat _change_type jsou nastavena na insert a delete události. Změny primárních klíčů mohou nastat, když se provádí ruční aktualizace jednoho z klíčových polí prostřednictvím příkazu UPDATE nebo MERGE. U tabulek typu SCD 2 mohou nastat změny, když se pole __start_at změní, aby odráželo dřívější hodnotu počáteční sekvence.

Dotaz AUTO CDC určuje hodnoty primárního klíče, které se liší pro zpracování typu SCD 1 a SCD typu 2:

Typ SCD Primární klíč
SCD typ 1, a rozhraní Python pipelines Primární klíč je hodnota keys parametru create_auto_cdc_flow() ve funkci. Pro rozhraní SQL je primární klíč sloupce definované KEYS klauzulí v AUTO CDC ... INTO příkazu.
SCD – typ 2 Primárním klíčem je keys parametr nebo KEYS klauzule plus návratová hodnota z coalesce(__START_AT, __END_AT) operace, kde __START_AT a __END_AT jsou odpovídajícími sloupci z cílové tabulky streamování. __START_AT Používá se, pokud je k dispozici a __END_AT kdy __START_AT je null (například počáteční záznam).

Přečtěte feed změn dat z materializovaného pohledu

Important

Tato funkce je v beta verzi.

Změnový datový tok můžete číst z materializovaného pohledu vytvořeného v Lakeflow pipeline nebo v Databricks SQL. Použijte to k replikaci změn materializovaných zobrazení na cílech mimo Azure Databricks, nebo k vedení historie změn materializovaných zobrazení pro audit a reportování.

Materializované pohledy používají automatický kanál změn dat, takže samotný kanál změn dat nemusíte zapínat. Místo toho aktivujete feed změn dat na každém materializovaném pohledu, na kterém ho potřebujete, splněním následujících požadavků. Viz Automatické změny datového kanálu.

  • Ke čtení kanálu změn dat musíte použít Databricks Runtime 18 LTS nebo novější v klasických výpočetních prostředích, v bezserverových výpočetních prostředích nebo v Databricks SQL.

  • Materializovaný pohled, pipeline, která jej vytváří, nebo pipeline, která jej čte, musí používat kanál PREVIEW.

  • Materializovaný pohled musí mít zapnuté sledování řádků. Materializované pohledy v bezserverových výpočetních prostředcích mají ve výchozím nastavení povolené sledování řádků. Viz sledování řádků v Azure Databricks. Chcete-li zkontrolovat, zda je sledování řádků povoleno v materializovaném zobrazení, spusťte:

    SHOW TBLPROPERTIES my_mv ('delta.enableRowTracking');

  • Chcete-li číst kanál změn dat z materializovaného pohledu, povolte v kanálu nebo v materializovaném pohledu příznak externích metadat. Instrukce najdete v článku Jak povolit přístup pro datovou sadu.

Kanál změn dat z materializovaného zobrazení čtete stejně jako z jiných tabulek Delta, pomocí funkce table_changes(), streamovacího čtení nebo možnosti readChangeFeed. Pro syntaxi a příklady v jazycích SQL a Python viz Použití kanálu změn dat v Azure Databricks.

Kanál změnových dat materializovaného pohledu můžete číst v materializovaném pohledu Databricks SQL nebo ve streamovací tabulce:

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

Limitations

Kromě omezení automatického kanálu dat změn platí při čtení kanálu dat změn z materializovaného zobrazení následující:

  • Feed změn obsahuje nezměněné řádky, když je materializovaný pohled plně přepsán, a nekonsoliduje více aktualizací stejného řádku do jedné události. Chcete-li je odfiltrovat, agregujte kanál změnových dat seskupením podle všech sloupců, abyste nalezli vložení a odstranění, která mají stejné hodnoty řádků.
  • Pouze Azure Databricks může dotazovat informační kanál změn dat pro materializované zobrazení. Externí klienti Delta Lake a Iceberg nemohou.
  • V rámci kanálů Lakeflow můžete kanál změnových dat materializovaného pohledu číst pouze z jiného kanálu a tento kanál musí používat kanál PREVIEW. Čtení datového toku změn materializovaného pohledu ve stejném pipeline, který jej vytváře, není podporováno.
  • Nelze vytvořit vektorový index vyhledávání z materializovaného pohledu.

Získání dat o záznamech zpracovaných CDC dotazem v pipelinech

Poznámka:

Následující metriky jsou zachyceny pouze AUTO CDC dotazy, nikoli AUTO CDC FROM SNAPSHOT dotazy.

Následující metriky jsou získávány pomocí dotazů AUTO CDC.

  • num_upserted_rows: Počet výstupních řádků přenesených do datové sady během aktualizace.
  • num_deleted_rows: Počet existujících výstupních řádků odstraněných z datové sady během aktualizace.

Metrika num_output_rows, výstup pro toky bez CDC, se pro dotazy nezachytává AUTO CDC.

Instalace částečných aktualizací

Pokud zdroj odešle pouze sloupce, které se změnily, musí rozlišovat mezi sloupcem, AUTO CDC který chybí v záznamu změn, který by měl ponechat cílovou hodnotu beze změny, a sloupec, který je explicitně nastaven na null, který by měl přepsat cílovou hodnotu null. Ve výchozím nastavení považuje IGNORE NULL UPDATES každé null za značku „neaktualizovat“, takže nemůže použít explicitní null. Pokud chcete tuto nejednoznačnost vyřešit, zvolte jednu z následujících tří metod:

Metoda Kdy ho použít Chování
IGNORE NULL UPDATES ON columnList Malá pevná sada sloupců by měla ignorovat null hodnoty, zatímco všechny ostatní sloupce používají explicitní null hodnoty. Uvedené sloupce zachovávají svou stávající cílovou hodnotu, pokud je nullpříchozí hodnota . Všechny ostatní sloupce používají explicitní null hodnoty.
IGNORE NULL UPDATES ON * EXCEPT (exceptColumnList) Většina sloupců by měla ignorovat null hodnoty a pouze několik z nich by mělo použít explicitní null hodnoty. Uvedené sloupce používají explicitní null hodnoty. Všechny ostatní sloupce zachovávají stávající cílovou hodnotu, pokud je nullpříchozí hodnota .
COLUMNS TO UPDATE Každý záznam změn aktualizuje jinou sadu sloupců nebo se v průběhu času změní sada aktualizovatelných sloupců. Zdrojový sloupec pojmenuje sloupce, které se mají aktualizovat pro každý záznam změn. Uvedené sloupce se zapisují ze zdroje, včetně explicitních null hodnot. Sloupce, které nejsou uvedené, zachovávají svou stávající cílovou hodnotu.

COLUMNS TO UPDATE nelze kombinovat s IGNORE NULL UPDATES a není podporován pro bitemporální tabulky.

Obecně platí, že zvolte COLUMNS TO UPDATE , kdy producent ví, které sloupce se změnily v každém záznamu, a může tyto informace přenášet ve zdrojovém sloupci, například když více producentů zapisuje do stejného zdroje nebo sada aktualizovatelných sloupců v průběhu času roste. Zvolte IGNORE NULL UPDATES ON , kdy vlastník kanálu předem zná pevnou sadu aktualizovatelných sloupců a dává přednost řízení v kódu kanálu.

Následující příklad používá zdrojový sloupec s názvem columnsToUpdate k určení toho, které sloupce jednotlivé záznamy změn aktualizují, včetně sloupců explicitně nastavených na null:

Python

from pyspark import pipelines as dp

dp.create_streaming_table("target")

dp.create_auto_cdc_flow(
  target = "target",
  source = "cdc_source",
  keys = ["id"],
  sequence_by = "sequenceNum",
  stored_as_scd_type = 1,
  columns_to_update = "columnsToUpdate"
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  target
FROM
  stream(cdc_source)
KEYS
  (id)
SEQUENCE BY
  sequenceNum
STORED AS
  SCD TYPE 1
COLUMNS TO UPDATE
  columnsToUpdate;

Úplný přehled parametrů viz AUTO CDC INTO (pipelines) a create_auto_cdc_flow.

Bitemporální 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.

Příklady pro Bitemporal AUTO CDC

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í posloupnost změn ukazuje, jak bitemporální tabulka zaznamenává vložení, aktualizaci, opožděnou aktualizaci a odstranění u jedné společnosti. 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.

Změna 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

Output:

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.

Změna 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

Output:

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.

Změna 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

Output:

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.

Změna 4: Odstranit

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

Output:

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.

Jaké datové objekty se používají ke zpracování CDC v datovém toku?

Když deklarujete cílovou tabulku v metastoru Hive, vytvoří se dvě datové struktury:

  • Zobrazení s názvem přiřazeným k cílové tabulce.
  • Interní zálohovací tabulka používaná kanálem ke správě zpracování CDC. Tato tabulka je pojmenovaná tak, že se předsadí __apply_changes_storage_ na název cílové tabulky.

Pokud například deklarujete cílovou tabulku s názvem dp_cdc_target, zobrazí se zobrazení s názvem dp_cdc_target a tabulka pojmenovaná __apply_changes_storage_dp_cdc_target v metastoru. Zadejte dotaz na zobrazení pro přístup ke zpracovaným datům. Neupravujte záložní tabulku přímo.

Poznámka:

Tyto datové struktury se vztahují pouze na AUTO CDC zpracování, nikoli AUTO CDC FROM SNAPSHOT zpracování. Platí také pouze pro metastor Hive, nikoli pro katalog Unity.

  • Last updated on