Rozhraní API APPLY CHANGES: Zjednodušení zachytávání změnových dat pomocí Delta Live Tables

Delta Live Tables zjednodušuje zachytávání dat změn (CDC) pomocí rozhraní API APPLY CHANGES a APPLY CHANGES FROM SNAPSHOT. Rozhraní, které používáte, závisí na zdroji změn dat:

• Slouží APPLY CHANGES ke zpracování změn z datového kanálu změn (CDF).
• Ke zpracování změn snímků databáze použijte APPLY CHANGES FROM SNAPSHOT (Public Preview).

Dříve se MERGE INTO tento příkaz běžně používal ke zpracování záznamů CDC v Azure Databricks. Může však MERGE INTO vést k nesprávným výsledkům z důvodu záznamů mimo posloupnosti nebo vyžaduje složitou logiku pro opakované řazení záznamů.

Rozhraní APPLY CHANGES API je podporováno v SQL a Python rozhraních pro Delta Live Tables. Rozhraní APPLY CHANGES FROM SNAPSHOT API se podporuje v rozhraní Pythonu Delta Live Tables.

APPLY CHANGES i APPLY CHANGES FROM SNAPSHOT podporují aktualizaci tabulek pomocí scd typu 1 a typu 2:

• Pomocí scd typu 1 aktualizujte záznamy přímo. Historie se neuchovává pro aktualizované záznamy.
• Pomocí scd typu 2 zachovejte historii záznamů, a to buď u všech aktualizací, nebo aktualizací v zadané sadě sloupců.

Syntaxi a další odkazy najdete tady:

• Snímání změn dat z kanálu změn pomocí Pythonu v Delta Live Tables
• Snímání změn dat pomocí SQL v Delta Live Tables

Poznámka:

Tento článek popisuje, jak aktualizovat tabulky v kanálu Delta Live Tables 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 Delta Lake 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 DLT nebo edicí Delta Live Tables Pro nebo Advancededice.

Jak je CDC implementováno s rozhraním APPLY CHANGES API?

Díky automatickému zpracování neřazených záznamů zajišťuje rozhraní API APPLY CHANGES v Delta Live Tables správné zpracování záznamů CDC a eliminuje potřebu vyvíjet složitou logiku pro zpracování neřazených záznamů. Ve zdrojových datech je nutné zadat sloupec, na kterém se mají sekvencovat záznamy, které Delta Live Tables interpretuje jako monotonicky rostoucí reprezentaci správného pořadí zdrojových dat. Delta Live Tables automaticky zpracovává data, která přicházejí mimo pořadí. U změn typu SCD 2 Delta Live Tables propaguje odpovídající hodnoty sekvencí do sloupců __START_AT a __END_AT cílové tabulky. V každé hodnotě sekvencování by měla existovat jedna samostatná aktualizace klíče a hodnoty sekvencování NULL nejsou podporovány.

Pokud chcete provádět zpracování CDC pomocí APPLY CHANGES, nejprve vytvoříte streamovací tabulku a pak použijete příkaz APPLY CHANGES INTO v SQL nebo funkci apply_changes() v Pythonu k určení zdroje, klíčů a sekvencování kanálu změn. Pokud chcete vytvořit cílovou streamovací tabulku, použijte příkaz CREATE OR REFRESH STREAMING TABLE v SQL nebo funkci create_streaming_table() v Pythonu. Podívejte se na příklady zpracování typu SCD typu 1 a typu 2.

Podrobnosti o syntaxi najdete v referenční příručce pro SQL Delta Live Tables nebo v referenční příručce pro Python .

Jak je CDC implementováno s rozhraním APPLY CHANGES FROM SNAPSHOT API?

Důležité

Rozhraní APPLY CHANGES FROM SNAPSHOT API je ve verzi Public Preview.

APPLY CHANGES FROM SNAPSHOT je deklarativní rozhraní API, které efektivně určuje změny ve zdrojových datech porovnáním řady snímků v daném pořadí a pak spustí zpracování potřebné ke zpracování záznamů ve snímcích CDC. APPLY CHANGES FROM SNAPSHOT podporuje pouze rozhraní Delta Live Tables Python.

APPLY CHANGES FROM SNAPSHOT podporuje ingestování snímků z více zdrojových typů:

• K příjmu snímků z existující tabulky nebo zobrazení použijte pravidelné příjem dat snímků. APPLY CHANGES FROM SNAPSHOT má jednoduché zjednodušené rozhraní, které podporuje pravidelné ingestování snímků z existujícího databázového objektu. Při každé aktualizaci pipeline se ingestuje nový snímek a jako verze snímku se použije čas příjmu dat. Při spuštění kanálu v průběžném režimu se ingestuje více snímků s každou aktualizací kanálu v období určeném nastavením intervalu aktivační události pro tok, který obsahuje zpracování SNÍMKU APPLY CHANGES FROM SNAPSHOT.
• K zpracování souborů obsahujících snímky databáze, jako jsou snímky vygenerované z databáze Oracle nebo MySQL nebo datového skladu, použijte historický příjem snímků.

Pokud chcete provádět zpracování CDC z libovolného typu zdroje pomocí APPLY CHANGES FROM SNAPSHOT, nejprve vytvoříte streamovací tabulku a pak pomocí funkce apply_changes_from_snapshot() v Pythonu určíte snímek, klíče a další argumenty potřebné k implementaci zpracování. Podívejte se na příklady pravidelného příjmu snímků a historického příjmu snímků.

Snímky předané rozhraní API musí být ve vzestupném pořadí podle verze. Pokud Delta Live Tables zjistí snímek, který není v pořadí, vyvolá se chyba.

Podrobnosti o syntaxi najdete v referenčníchpythonových tabulek delta live.

Omezení

Sloupec použitý pro sekvencování musí být seřaditelný datový typ.

Příklad: Zpracování SCD typu 1 a SCD typu 2 se zdrojovými daty CDF

Následující části poskytují příklady dotazů typu 1 a typu 2 SCD pro Delta Live Tables, které aktualizují cílové tabulky na základě zdrojových událostí z datového kanálu změn, které:

1. Vytvoří nové záznamy uživatelů.
2. Odstraní záznam uživatele.
3. Aktualizuje záznamy uživatelů. V příkladu typu SCD 1 se poslední operace UPDATE zpozdí a vyřadí z cílové tabulky, což demonstruje zpracování událostí, které nejsou ve správném pořadí.

Následující příklady předpokládají znalost konfigurace a aktualizace kanálů Delta Live Tables. Viz návod : Spuštění prvního datového kanálu Delta Live Tables.

Abyste mohli tyto příklady spustit, musíte začít vytvořením ukázkové datové sady. Viz Generování testovacích dat.

Tady jsou vstupní záznamy pro tyto příklady:

userId	name	city	operation	sequenceNum
124	Raul	Oaxaca	INSERT	0
123	Isabel	Monterrey	INSERT	0
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 ukázkových datech, vloží se následující záznam, který určuje, kam se mají záznamy zkrátit:

userId	name	city	operation	sequenceNum
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ý.

Zpracování aktualizací typu SCD typu 1

Následující příklad ukazuje zpracování aktualizací typu SCD 1:

Python

import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def users():
  return spark.readStream.table("cdc_data.users")

dlt.create_streaming_table("target")

dlt.apply_changes(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

APPLY CHANGES INTO
  target
FROM
  stream(cdc_data.users)
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 SCD 1 obsahuje cílová tabulka následující záznamy:

userId	name	city
124	Raul	Oaxaca
125	Mercedes	Guadalajara
126	Lilie	Cancun

Po spuštění příkladu typu SCD 1 s dalším záznamem TRUNCATE se záznamy 124 a 126 zkrátí kvůli operaci TRUNCATE v sequenceNum=3a cílová tabulka obsahuje následující záznam:

userId	name	city
125	Mercedes	Guadalajara

Zpracování aktualizací typu SCD typu 2

Následující příklad ukazuje zpracování aktualizací typu 2 SCD:

Python

import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def users():
  return spark.readStream.table("cdc_data.users")

dlt.create_streaming_table("target")

dlt.apply_changes(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

APPLY CHANGES INTO
  target
FROM
  stream(cdc_data.users)
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	name	city	__START_AT	__END_AT
123	Isabel	Monterrey	0	5
123	Isabel	Chihuahua	5	6
124	Raul	Oaxaca	0	null
125	Mercedes	Tijuana	2	5
125	Mercedes	Mexicali	5	6
125	Mercedes	Guadalajara	6	null
126	Lilie	Cancun	2	null

Dotaz TYPU 2 SCD může také určit podmnožinu výstupních sloupců, které se mají sledovat pro historii v cílové tabulce. Změny v jiných sloupcích se aktualizují místo generování nových záznamů historie. Následující příklad ukazuje vyloučení sloupce city ze sledování:

Následující příklad ukazuje použití historie sledování s SCD typu 2:

Python

import dlt
from pyspark.sql.functions import col, expr

@dlt.view
def users():
  return spark.readStream.table("cdc_data.users")

dlt.create_streaming_table("target")

dlt.apply_changes(
  target = "target",
  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 and populate the target table.
CREATE OR REFRESH STREAMING TABLE target;

APPLY CHANGES INTO
  target
FROM
  stream(cdc_data.users)
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)

Po spuštění tohoto příkladu bez dalšího záznamu TRUNCATE obsahuje cílová tabulka následující záznamy:

userId	name	city	__START_AT	__END_AT
123	Isabel	Chihuahua	0	6
124	Raul	Oaxaca	0	null
125	Mercedes	Guadalajara	2	null
126	Lilie	Cancun	2	null

Generování testovacích dat

Níže uvedený kód slouží k vygenerování ukázkové datové sady pro použití v ukázkových dotazech, které jsou přítomné v tomto kurzu. Za předpokladu, že máte správné přihlašovací údaje k vytvoření nového schématu a vytvoření nové tabulky, můžete tyto příkazy spustit pomocí poznámkového bloku nebo Databricks SQL. Následující kód je není určený ke spuštění jako součást kanálu Delta Live Tables:

CREATE SCHEMA IF NOT EXISTS cdc_data;

CREATE TABLE
  cdc_data.users
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 above batch at sequenceNum 6 will be the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Příklad: Pravidelné zpracování snímků

Následující příklad ukazuje zpracování typu SCD 2, které zpracovává snímky tabulky uložené v mycatalog.myschema.mytable. Výsledky zpracování se zapisují do tabulky s názvem target.

mycatalog.myschema.mytable záznamy v časovém razítku 2024-01-01 00:00:00

Key	Hodnota
0	a1
2	a2

mycatalog.myschema.mytable záznamy v časovém razítku 2024-01-01 12:00:00

Key	Hodnota
2	b2
3	a3

import dlt

@dlt.view(name="source")
def source():
 return spark.read.table("mycatalog.myschema.mytable")

dlt.create_streaming_table("target")

dlt.apply_changes_from_snapshot(
 target="target",
 source="source",
 keys=["key"],
 stored_as_scd_type=2
)

Po zpracování snímků obsahuje cílová tabulka následující záznamy:

Key	Hodnota	__START_AT	__END_AT
0	a1	2024-01-01 00:00:00	2024-01-01 12:00:00
2	a2	2024-01-01 00:00:00	2024-01-01 12:00:00
2	b2	2024-01-01 12:00:00	null
3	a3	2024-01-01 12:00:00	null

Příklad: Historické zpracování snímků

Následující příklad ukazuje zpracování typu SCD 2, které aktualizuje cílovou tabulku na základě zdrojových událostí ze dvou snímků uložených v systému cloudového úložiště:

Snímek v timestamp, uložený v /<PATH>/filename1.csv

Key	TrackingColumn	NonTrackingColumn
0	a1	b1
2	a2	b2
4	a4	b4

Snímek v timestamp + 5, uložený v /<PATH>/filename2.csv

Key	TrackingColumn	NonTrackingColumn
2	a2_new	b2
3	a3	b3
4	a4	b4_new

Následující příklad kódu ukazuje zpracování aktualizací SCD typu 2 pomocí těchto snímků:

import dlt

def exist(file_name):
  # Storage system-dependent function that returns true if file_name exists, false otherwise

# This function returns a tuple, where the first value is a DataFrame containing the snapshot
# records to process, and the second value is the snapshot version representing the logical
# order of the snapshot.
# Returns None if no snapshot exists.
def next_snapshot_and_version(latest_snapshot_version):
  latest_snapshot_version = latest_snapshot_version or 0
  next_version = latest_snapshot_version + 1
  file_name = "dir_path/filename_" + next_version + ".csv"
  if (exist(file_name)):
    return (spark.read.load(file_name), next_version)
   else:
     # No snapshot available
     return None

dlt.create_streaming_live_table("target")

dlt.apply_changes_from_snapshot(
  target = "target",
  source = next_snapshot_and_version,
  keys = ["Key"],
  stored_as_scd_type = 2,
  track_history_column_list = ["TrackingCol"]
)

Po zpracování snímků obsahuje cílová tabulka následující záznamy:

Key	TrackingColumn	NonTrackingColumn	__START_AT	__END_AT
0	a1	b1	0	2
2	a2	b2	0	2
2	a2_new	b2	2	null
3	a3	b3	2	null
4	a4	b4_new	0	null

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

Pokud váš datový kanál publikuje tabulky do Unity katalogu, můžete použít příkazy jazyka pro manipulaci s daty (DML), včetně příkazů INSERT, UPDATE, DELETE a MERGE, k úpravě cílových streamovaných tabulek vytvořených příkazy APPLY CHANGES 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, je možné spustit pouze ve sdíleném clusteru Unity Catalog nebo SQL Warehouse při použití Databricks Runtime 13.3 LTS a novější.
• Vzhledem k tomu, že streamování vyžaduje zdroje dat pouze pro přidávání, pokud vaše zpracování vyžaduje streamování ze zdrojové streamovací tabulky, která obsahuje změny (například příkazy DML), nastavte příznak skipChangeCommits při čtení této 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 tabulku streamování, můžete jako cílovou tabulku použít materializované zobrazení (které nemá omezení pouze pro přidávání).

Vzhledem k tomu, že Delta Live Tables 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 typ 2), je nutné zajistit, aby příkazy DML používaly platné hodnoty pro tyto sloupce, aby zachovaly správné pořadí záznamů. Podívejte se , jak je CDC implementováno pomocí rozhraní API APPLY CHANGES?.

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

čtení datového kanálu změn z cílové tabulky APPLY CHANGES

Ve službě Databricks Runtime 15.2 a vyšší můžete číst kanál změnových dat ze streamovací tabulky, která je určená jako cíl APPLY CHANGES nebo APPLY CHANGES FROM SNAPSHOT dotazů, stejným způsobem, jako když čtete kanál změnových dat z jiných tabulek Delta. K přečtení změnového datového proudu z cílové streamovací tabulky jsou vyžadovány následující věci:

• Cílová streamovací tabulka musí být publikována v katalogu Unity. Viz Použijte katalog Unity s pipelinami Delta Live Tables.
• 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 kanálu Delta Live Tables, musí být kanál nakonfigurovaný tak, aby používal Databricks Runtime 15.2 nebo vyšší.

Datový kanál změn si přečtete z cílové streamovací tabulky vytvořené v kanálu Delta Live Tables stejným způsobem jako čtení datového kanálu změn z jiných tabulek Delta. Další informace o používání funkcí kanálu rozdílových změn dat, včetně příkladů v Pythonu a SQL, najdete v tématu Použití datového kanálu změn Delta Lake v 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.

** Hodnoty _change_type se ale liší, pokud dojde k aktualizaci cílové streamovací tabulky, která zahrnuje změnu hodnot primárního klíče. Pokud změny zahrnují aktualizace primárních klíčů, jsou pole metadat _change_type nastavena na insert a delete události. Ke změnám primárních klíčů může dojít v případě, že se ruční aktualizace pro jedno z polí klíčů provádějí pomocí příkazu UPDATE nebo MERGE nebo v případě tabulek typu SCD typu 2, když se pole __start_at změní tak, aby odráželo dřívější počáteční hodnotu sekvence.

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

• Pro zpracování typu SCD typu 1 a rozhraní Pythonu Delta Live Tables je primárním klíčem hodnota parametru keys ve funkci apply_changes(). Pro rozhraní SQL Delta Live Tables je primárním klíčem sloupce definované klauzulí KEYS v příkazu APPLY CHANGES INTO.
• U SCD typu 2 je primárním klíčem keys parametr nebo klauzule KEYS plus návratová hodnota z operace coalesce(__START_AT, __END_AT), kde __START_AT a __END_AT jsou odpovídající sloupce z cílové tabulky streamování.

Získání dat o záznamech zpracovaných dotazem CDC delta live tables

Poznámka:

Následující metriky jsou zachyceny pouze APPLY CHANGES dotazy, nikoli APPLY CHANGES FROM SNAPSHOT dotazy.

Dotazy zaznamenávají APPLY CHANGES následující metriky:

• 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á apply changes .

Jaké datové objekty se používají pro zpracování CDC v Delta Live Tables?

Poznámka:

• Tyto datové struktury se vztahují pouze na zpracování APPLY CHANGES, nikoli na zpracování APPLY CHANGES FROM SNAPSHOT.
• Tyto datové struktury platí pouze v případech, kdy je cílová tabulka publikovaná v metastoru Hive. Pokud pipelina publikuje do katalogu Unity, jsou interní podpůrné tabulky pro uživatele nepřístupné.

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í podpůrná tabulka používaná Delta Live Tables pro správu zpracování CDC. Tato tabulka je pojmenována připojením __apply_changes_storage_ před názvem cílové tabulky.

Pokud například deklarujete cílovou tabulku s názvem dlt_cdc_target, zobrazí se v metastoru zobrazení s názvem dlt_cdc_target a tabulka s názvem __apply_changes_storage_dlt_cdc_target. Vytvoření zobrazení umožňuje Delta Live Tables odfiltrovat dodatečné informace (například náhrobky a verze) nezbytné pro zpracování nepořádně uspořádaných dat. Pokud chcete zobrazit zpracovávaná data, zadejte dotaz na cílové zobrazení. Vzhledem k tomu, že schéma tabulky __apply_changes_storage_ se může změnit tak, aby podporovalo budoucí funkce nebo vylepšení, neměli byste se dotazovat na tabulku pro produkční použití. Pokud do tabulky přidáte data ručně, předpokládá se, že záznamy přicházejí před dalšími změnami, protože sloupce verze chybí.