APPLY CHANGES API: Zjednodušení zachytávání dat změn v rozdílových živých tabulkách

Delta Live Tables zjednodušuje zachytávání dat změn (CDC) pomocí APPLY CHANGES rozhraní API. 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 kvůli zastaralým záznamům nebo k opětovnému řazení záznamů vyžaduje složitou logiku.

Díky automatickému zpracování zastaralých záznamů APPLY CHANGES zajišťuje rozhraní API v rozdílových živých tabulkách správné zpracování záznamů CDC a eliminuje potřebu vyvíjet složitou logiku pro zpracování záznamů mimo posloupnosti.

Rozhraní APPLY CHANGES API je podporováno v rozhraníCH SQL a Pythonu pro dynamické tabulky Delta, včetně podpory pro 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:

• Změna zachytávání dat pomocí Pythonu v rozdílových živých tabulkách
• Změna zachytávání dat pomocí SQL v rozdílových živých tabulkách
• Řízení správy náhrobků pro dotazy typu 1 typu SCD

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.

Jak je CDC implementováno s rozdílovými živými tabulkami?

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 2 typu SCD rozšíří rozdílové živé tabulky příslušné hodnoty sekvencování do __START_AT sloupců __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 s rozdílovými živými tabulkami, nejprve vytvoříte streamovací tabulku a pak pomocí APPLY CHANGES INTO příkazu určíte zdroj, klíče a sekvencování kanálu změn. Pokud chcete vytvořit cílovou streamovací tabulku, použijte CREATE OR REFRESH STREAMING TABLE příkaz v SQL nebo funkci v Pythonu create_streaming_table() . Pokud chcete vytvořit příkaz definující zpracování CDC, použijte příkaz APPLY CHANGES v SQL nebo funkci v Pythonu apply_changes() . Podrobnosti o syntaxi najdete v tématu Změna zachytávání dat pomocí SQL v rozdílových živých tabulkách nebo změna zachytávání dat pomocí Pythonu v rozdílových živých tabulkách.

Jaké datové objekty se používají ke zpracování CDC delta živých tabulek?

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á delta živými tabulkami 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 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 rozdílovým živým tabulkám vyfiltrovat dodatečné informace (například náhrobky a verze) potřebné ke zpracování dat mimo pořadí. Pokud chcete zobrazit zpracovávaná data, zadejte dotaz na cílové zobrazení. Vzhledem k tomu, že schéma __apply_changes_storage_ tabulky se může změnit tak, aby podporovalo budoucí funkce nebo vylepšení, neměli byste se dotazovat na tabulku pro použití v produkčním prostředí. 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í.

Pokud kanál publikuje do katalogu Unity, nebudou interní záložní tabulky uživatelům přístupné.

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

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 , která je výstupem pro toky mimo CDC, se pro apply changes dotazy nezachytává.

Omezení

Cíl APPLY CHANGES INTO dotazu nebo apply_changes funkce nelze použít jako zdroj pro tabulku streamování. Tabulka, která čte z cíle APPLY CHANGES INTO dotazu nebo apply_changes funkce, musí být materializované zobrazení.

SCD – typ 1 a SCD – typ 2 v Azure Databricks

Následující části obsahují příklady, které demonstrují SCD typu 1 a 2 dotazů, které aktualizují cílové tabulky na základě zdrojových událostí, které:

1. Vytvořte nové záznamy uživatelů.
2. Odstranění záznamu uživatele
3. Aktualizujte záznamy uživatelů. V příkladu SCD typu 1 se poslední UPDATE operace zpozdí a vyřadí se z cílové tabulky, což demonstruje zpracování událostí mimo pořadí.

Následující příklady předpokládají znalost konfigurace a aktualizace kanálů Delta Live Tables. Viz kurz: Spuštění prvního kanálu dynamických tabulek Delta.

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	Lily	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 zadání obou DELETETRUNCATE operací, ale každá z těchto možností je volitelná.

Zpracování aktualizací typu SCD typu 1

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

Python

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

@dlt.view
def users():
  return spark.readStream.format("delta").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
  live.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	Lily	Cancun

Po spuštění příkladu SCD typu 1 s dalším TRUNCATE záznamem se záznamy 124 a 126 jsou zkráceny z důvodu TRUNCATE operace at 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 kódu ukazuje zpracování aktualizací TYPU 2 SCD:

Python

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

@dlt.view
def users():
  return spark.readStream.format("delta").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
  live.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	1	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	Lily	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í city sloupce 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.format("delta").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
  live.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 TRUNCATE záznamu obsahuje cílová tabulka následující záznamy:

userId	name	city	__START_AT	__END_AT
123	Isabel	Chihuahua	1	6
124	Raul	Oaxaca	0	null
125	Mercedes	Guadalajara	2	null
126	Lily	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 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 5 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ř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 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 Katalogu Unity nebo SQL Warehouse pomocí databricks Runtime 13.3 LTS a vyšší.
• Vzhledem k tomu, že streamování vyžaduje zdroje dat jen pro připojení, pokud vaše zpracování vyžaduje streamování ze zdrojové tabulky streamování se změnami (například příkazy DML), nastavte příznak skipChangeCommits při čtení zdrojové tabulky streamování. Při skipChangeCommits nastavení se transakce, které odstraňují nebo upravují záznamy ve zdrojové tabulce, ignorují. Pokud zpracování nevyžaduje tabulku streamování, můžete jako cílovou tabulku použít materializované zobrazení (které nemá omezení pouze pro připojení).

Vzhledem k tomu, že tabulky Delta Live Tables používají zadaný SEQUENCE BY sloupec a šíří odpovídající hodnoty sekvencování do __START_AT a __END_AT sloupců cílové tabulky (pro SCD typu 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 je CDC implementováno s rozdílovými živými tabulkami?.

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