Použití kanálu změn dat na Azure Databricks

Tok dat o změnách (CDF) sleduje změny na úrovni řádků mezi verzemi tabulky Delta Lake nebo tabulky Apache Iceberg v3.

Azure Databricks podporuje dva přístupy:

  • Automatický datový kanál změn: Vypočítá změny během čtení tabulky pomocí metadat rodokmenu řádků. Nevyžaduje konfiguraci jednotlivých tabulek a funguje u tabulek Delta Lake a Apache Iceberg v3. Viz Automatické změny datového kanálu.
  • Datový kanál starších verzí změn: Materializuje změny během zápisů tabulek. Podporuje pouze tabulky Delta Lake. Vyžaduje konfiguraci jednotlivých tabulek. Viz Starší kanál změnových dat pro Delta Lake.

Kanál změn dat můžete použít pro běžné případy použití dat, mezi které patří:

  • Přírůstkové kanály ETL, které zpracovávají pouze řádky, které se od posledního spuštění kanálu změnily.
  • Záznamy auditu, které sledují úpravy dat pro požadavky na dodržování předpisů a zásady správného řízení.
  • Úlohy replikace dat, které synchronizují změny podřízených tabulek, mezipamětí nebo externích systémů

Automatické změny datového kanálu

Important

Tato funkce je ve verzi Public Preview. Správci pracovního prostoru můžou řídit přístup k této funkci ze stránky Previews . Viz Manage Azure Databricks preview.

Automatický kanál změn dat vypočítává změny na úrovni řádků při dotazu, nikoli při zápisu, pomocí sledování řádků v Delta Lake a lineage řádků v Apache Iceberg v3. Na rozdíl od staršího kanálu změn dat automatický kanál změn dat nevyžaduje konfiguraci jednotlivých tabulek a funguje s tabulkami Delta Lake a Apache Iceberg v3.

Protože se změny nepočítají při každém zápisu u operací MERGE INTO a UPDATE, automatické kanály změn dat ve srovnání s původními kanály změn dat zvyšují výkon zápisu a snižují náklady na úložiště.

Automatický kanál dat o změnách používá stejná rozhraní API table_changes() a readChangeFeed jako starší kanál dat o změnách a funguje s dávkovými dotazy, Structured Streaming a sdílením Delta Lake z Databricksu do Databricksu. Viz Čtení změn pomocí dávkových dotazů a Přírůstkové zpracování dat o změnách.

Požadavky

  • Databricks Runtime 18 nebo novější
  • Podporovaný formát tabulky zaregistrovaný v katalogu Unity:
    • Spravovaná tabulka ve formátu Delta Lake s povoleným sledováním řádků nebo ve formátu Iceberg v3
    • Externí tabulka ve formátu Delta Lake s povoleným sledováním řádků

Viz typy tabulek katalogu Databricks Unity.

Poznámka:

Změna datového kanálu není součástí specifikace Apache Iceberg. Azure Databricks čtenáři můžou zadávat dotazy na automatické datové kanály změn pro tabulky Apache Iceberg v3, ale externí čtenáři Icebergu nemůžou. Podívejte se na specifikaci tabulky Iceberg.

Pro Delta Lake mohou automatický kanál změn dat dotazovat pouze čtecí klienti Azure Databricks.

Použití kanálu pro změny dat

Pokud chcete použít datový kanál změn, ověřte, že používáte tabulku, která splňuje požadavky. Viz Požadavky.

Pokud chcete dávkově číst datový kanál změn, postupujte takto:

Python

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("<table_name>")

Scala

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("<table_name>")

SQL

SELECT * FROM table_changes('<table_name>', 0)

Další informace o dávkovém čtení z kanálu změn dat najdete v tématu Čtení změn v dávkových dotazech.

Chcete-li číst kanál dat o změnách pomocí streamu, postupujte takto:

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("<table_name>")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("<table_name>")

Další informace o čtení ze streamu v kanálu změn dat najdete v tématu Přírůstkové zpracování změnových dat.

Migrace ze staršího kanálu dat o změnách

Pokud chcete migrovat tabulku Delta Lake ze starší verze datového kanálu změn na automatický datový kanál změn, postupujte takto:

  1. Ověřte, že vaše tabulka splňuje požadavky.
  2. Vypněte starší datový kanál změn spuštěním následujícího příkazu:
ALTER TABLE <table_name> UNSET TBLPROPERTIES ('delta.enableChangeDataFeed');

Starší i automatické datové kanály změn nemůžete používat společně.

Změna schématu datového kanálu

Při čtení z datového kanálu změn pro tabulku použije dotaz schéma pro nejnovější verzi tabulky. Azure Databricks podporuje většinu operací změny a vývoje schématu, ale tabulky s mapováním sloupců mají omezení. Viz Tabulky s mapováním sloupců.

Kromě sloupců dat ze schématu tabulky Delta Lake obsahuje kanál dat změny sloupce metadat, které identifikují typ události změny:

Název sloupce Typ Hodnoty
_change_type String Obsahuje: insert, update_preimage, update_postimage, delete.
preimage je hodnota před aktualizací, postimage je hodnota za aktualizací.
_commit_version Long Obsahuje: verze protokolu Delta nebo tabulky obsahující změnu.
_commit_timestamp Časové razítko Obsahuje: časové razítko přidružené k vytvoření potvrzení.

Pokud schéma obsahuje sloupce se stejnými názvy jako tyto sloupce metadat, nemůžete v tabulce použít datový kanál změn. Než zapnete funkci Change Data Feed, přejmenujte sloupce v tabulce, abyste tento konflikt vyřešili.

Postupné zpracování změnových dat

Databricks doporučuje používat kanál změn dat v kombinaci se strukturovaným streamováním k postupnému zpracování změn z tabulek. Strukturované streamování pro Azure Databricks musíte použít k automatickému sledování verzí pro datový kanál změn vaší tabulky. Zpracování CDC s tabulkami SCD typu 1 nebo typu 2 najdete v tématu Automatická rozhraní API CDC: Zjednodušení zachytávání dat změn pomocí kanálů.

Když se stream poprvé spustí, kanál dat o změnách vrací nejnovější snímek tabulky ve formě záznamů INSERT a poté vrací budoucí změny jako data o změnách. Kanály změn dat současně zapisují změny dat i nově vložené řádky do transakčního protokolu tabulky.

Chcete-li nakonfigurovat datový tok tak, aby četl kanál změn dat tabulky, nastavte možnost readChangeFeed na true následovně:

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .table("myTable")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .table("myTable")

Omezení rychlosti

Azure Databricks podporuje omezení rychlosti (maxFilesPerTrigger, maxBytesPerTrigger) a excludeRegex při čtení dat o změnách. Úplný seznam možností streamování v Delta Lake najdete v tématu Delta Lake.

Volitelně můžete zadat počáteční verzi, viz Zadání počáteční verze. U verzí jiných než výchozí snímek se limity počtu požadavků vztahují atomicky na celé commity. Buď aktuální dávka obsahuje celý commit, nebo jeho zařazení odloží do další dávky.

Přehrání historie tabulek

Datový kanál změn není určen jako trvalý záznam všech změn v tabulce. Zaznamenává pouze změny, ke kterým dochází po povolení datového kanálu změn. Můžete zahájit nové čtení streamování, abyste zachytili aktuální verzi a všechny následné změny.

Záznamy v datovém kanálu změn jsou přechodné a přístupné pouze pro zadané okno uchovávání informací. Transakční protokoly v pravidelných intervalech odstraňují verze tabulek a jejich odpovídající verze change data feed. Po odebrání verze už nemůžete číst datový kanál změn pro danou verzi.

Archivovat data o změnách pro trvalé uchování historie

Pokud váš případ použití vyžaduje, abyste zachovali trvalou historii všech změn v tabulce, použijte přírůstkovou logiku k zápisu záznamů z datového kanálu změn do nové tabulky.

Následující příklad ukazuje použití trigger.AvailableNow ke zpracování dostupných dat jako dávkové úlohy pro audit nebo kompletní přehrání změn:

Python
(spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(availableNow=True)
  .toTable("target_table")
)
Scala
spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(Trigger.AvailableNow)
  .toTable("target_table")

Zadání počáteční verze

Pokud chcete číst změny z konkrétního bodu, zadejte počáteční verzi pomocí časového razítka nebo čísla verze. Pro dávkové čtení se vyžadují počáteční verze. Volitelně můžete zadat koncovou verzi pro omezení rozsahu. Další informace o historii tabulek najdete v tématu Časová cesta.

Když nakonfigurujete úlohy strukturovaného streamování, které používají datový kanál změn, může určení počáteční verze ovlivnit výkon zpracování:

  • Nové kanály zpracování dat obvykle využívají výchozí chování, které zaznamenává všechny existující záznamy v tabulce jako INSERT operace při prvním spuštění datového proudu.
  • Pokud cílová tabulka již obsahuje všechny záznamy s odpovídajícími změnami až do určitého bodu, zadejte počáteční verzi, aby se zabránilo zpracování stavu zdrojové tabulky jako INSERT událostí.

Následující příklad ukazuje, jak se zotavit z selhání streamování s poškozeným kontrolním bodem. V tomto příkladu předpokládejme následující podmínky:

  1. U zdrojové tabulky byla při vytváření tabulky povolená změna datového kanálu.
  2. Cílová navazující tabulka zpracovala všechny změny až po verzi 75 včetně.
  3. Historie verzí zdrojové tabulky je k dispozici pro verze 70 a vyšší.

Když definujete stream zápisu do existující cílové tabulky, musíte zadat nové umístění kontrolního bodu:

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
  .writeStream
  .option("checkpointLocation", "<new-checkpoint-path>")
  .toTable("target_table")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
  .writeStream
  .option("checkpointLocation", "<new-checkpoint-path>")
  .toTable("target_table")

Important

Pokud zadáte počáteční verzi a tato verze není v historii tabulek dostupná, stream se nespustí z nového kontrolního bodu. Vzhledem k tomu, že spravované tabulky automaticky vyčistí historické verze, všechny zadané počáteční verze se nakonec odstraní.

Viz Historii tabulky.

Čtení změn v dávkových dotazech

Syntaxi dávkového dotazu můžete použít ke čtení všech změn počínaje konkrétní verzí nebo ke čtení změn v zadaném rozsahu verzí následujícím způsobem:

  • Zadejte verze jako celá čísla a časová razítka jako řetězce ve formátu yyyy-MM-dd[ HH:mm:ss[.SSS]].
  • Počáteční a koncové verze jsou inkluzivní. Pokud chcete číst od počáteční verze po nejnovější verzi, zadejte pouze počáteční verzi.
  • Pokud zadáte verzi z doby předtím, než byl change data feed povolen, dojde k chybě.

Pokud chcete používat dávkové čtení s možnostmi počáteční a koncové verze, postupujte takto:

SQL

Pokud chcete číst z verze 0 do 10, postupujte takto:

SELECT * FROM table_changes('tableName', 0, 10)

Pokud chcete číst mezi dvěma verzemi časového razítka, postupujte takto:

--
SELECT * FROM table_changes('tableName', '2021-04-21 05:45:46', '2021-05-21 12:00:00')

Pokud chcete číst od počáteční verze k nejnovější verzi, postupujte takto:

SELECT * FROM table_changes('tableName', 0)

Pokud chcete přečíst změny tabulky se speciálními znaky v názvu, postupujte takto:

SELECT * FROM table_changes('`schema`.`dotted.tableName`', '2021-04-21 06:45:46', '2021-05-21 12:00:00')

Viz tabulkovou funkci table_changes.

Python

Pokud chcete číst z verze 0 do 10, postupujte takto:

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .option("endingVersion", 10) \
  .table("myDeltaTable")

Pokud chcete číst mezi dvěma časovými razítky, postupujte takto:

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingTimestamp", '2021-04-21 05:45:46') \
  .option("endingTimestamp", '2021-05-21 12:00:00') \
  .table("myDeltaTable")

Pokud chcete číst od počáteční verze k nejnovější verzi, postupujte takto:

spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("myDeltaTable")

Scala

Pokud chcete číst z verze 0 do 10, postupujte takto:

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .option("endingVersion", 10)
  .table("myDeltaTable")

Pokud chcete číst mezi dvěma časovými razítky, postupujte takto:

spark.read
  .option("readChangeFeed", "true")
  .option("startingTimestamp", "2021-04-21 05:45:46")
  .option("endingTimestamp", "2021-05-21 12:00:00")
  .table("myDeltaTable")

Pokud chcete číst od počáteční verze k nejnovější verzi, postupujte takto:

spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("myDeltaTable")

Zpracování verzí mimo rozsah

Pokud ve výchozím nastavení zadáte verzi nebo časové razítko, které překračuje poslední potvrzení, dotaz vrátí chybu timestampGreaterThanLatestCommit.

V Databricks Runtime 11.3 LTS a novějších verzích můžete povolit toleranci pro verze mimo rozsah následujícím způsobem:

SET spark.databricks.delta.changeDataFeed.timestampOutOfRange.enabled = true;

Pokud je tato konfigurace povolená, dotaz vrátí různé výsledky následujícím způsobem:

  • Počáteční verze nebo časové razítko nad rámec posledního potvrzení vrátí prázdný výsledek.
  • Koncová verze nebo časové razítko pozdější než poslední revize vrátí všechny změny od začátku až po poslední revizi.

Starší tok dat změn pro Delta Lake

Starší verze datového kanálu změn vyžaduje ruční konfiguraci pro jednotlivé tabulky Delta Lake. Vzhledem k tomu, že datový kanál změn není součástí specifikace Apache Iceberg, nejsou tabulky Apache Iceberg podporovány. Databricks doporučuje přejít na automatické podávání datových změn. Viz Migrace ze staršího kanálu změnových dat.

Když je zapnutý starší kanál změn dat, běhové prostředí zaznamenává události změn pro všechna data zapsaná do tabulky. To zahrnuje data řádků spolu s metadaty, která označují, jestli byl zadaný řádek vložen, odstraněn nebo aktualizován.

Starší kanál dat o změnách používá stejná rozhraní API readChangeFeed a table_changes() pro čtení jako automatický kanál dat o změnách. Viz Postupné zpracování dat o změnách a Čtení změn v dávkových dotazech.

Zapnutí datového kanálu starší verze změn

U jednotlivých tabulek musíte explicitně zapnout datový kanál starších změn. Použijte jednu z následujících metod:

Nová tabulka

Nastavte vlastnost delta.enableChangeDataFeed = true tabulky v CREATE TABLE příkazu.

CREATE TABLE student (id INT, name STRING, age INT)
  TBLPROPERTIES (delta.enableChangeDataFeed = true)

Poznámka:

Pokud starší kanál dat o změnách na určitou dobu vypnete a potom ho znovu zapnete, nebude možné daný interval dotazovat. K dotazování na změny v daném intervalu použijte automatický kanál dat o změnách. Viz Automatické změny datového kanálu.

Existující tabulka

Nastavte vlastnost delta.enableChangeDataFeed = true tabulky v ALTER TABLE příkazu.

ALTER TABLE myDeltaTable
  SET TBLPROPERTIES (delta.enableChangeDataFeed = true)

Důležité informace o úložišti

Spravované tabulky zaznamenávají změny dat efektivně a můžou k optimalizaci rozložení úložiště používat další funkce.

U staršího kanálu dat o změnách musíte vzít v úvahu následující chování při ukládání:

  • Může se zobrazit malý nárůst nákladů na úložiště, protože změny se můžou zaznamenávat v samostatných souborech.
  • Některé operace, například operace pouze s vkládáním nebo úplné smazání oddílů, negenerují soubory změnových dat. Azure Databricks vypočítá datový kanál změn přímo z transakčního protokolu.
  • Změňte datové soubory pomocí zásad uchovávání informací v tabulce. Příkaz VACUUM odstraňuje soubory s daty změn a na změny z transakčního protokolu se vztahují zásady uchovávání checkpointů.

Databricks doporučuje, abyste se nepokoušli rekonstruovat datový kanál změn přímým dotazováním datových souborů změn. Vždy používejte rozhraní Delta Lake a Apache Iceberg API.

Omezení

Zvažte následující omezení pro kanály dat o změnách:

Tabulky s mapováním sloupců

Když je u tabulky Delta Lake povolené mapování sloupců, můžete sloupce vypustit nebo přejmenovat bez přepsání datových souborů. Viz Přejmenování a vyřazení sloupců v rámci mapování sloupců Delta Lake.

Kanály datových změn však mají po neaditivních změnách schématu omezení. Neaditivní změny schématu zahrnují následující operace:

  • Přejmenujte nebo odstraňte sloupce.
  • Změnit datové typy sloupců
  • Změňte možnost hodnoty NULL u sloupce, například pomocí ALTER COLUMN ... SET NOT NULL. Viz NOT NULL omezení.

Kanály změnových dat nelze číst pro transakci nebo rozsah, ve kterém dojde k neaditivní změně schématu.

Aby bylo možné zohlednit neaditivní změny schématu před zadaným rozsahem dávkových čtení nebo po něm, dotazy používají schéma verze na konci rozsahu namísto nejnovější verze tabulky. Dotazy stále selžou, pokud rozsah verzí zahrnuje změnu schématu bez sčítání.

Automatické změny datového kanálu

  • Vzhledem k tomu, že v specifikaci Apache Iceberg není podporovaný datový kanál změn, externí klienti Icebergu se nemůžou dotazovat na automatické datové kanály změn. Podívejte se na specifikaci tabulky Iceberg.
  • U transakcí s více příkazy není automatické poskytování dat o změnách podporováno, pokud byla během transakce změněna zdrojová tabulka.
  • Automatické podávání dat o změnách není u tabulek s filtry řádků nebo maskami sloupců podporováno. Viz filtry řádků a masky sloupců.
  • Dotazy na kanál změn dat nemůžou zahrnovat verze tabulky, ve kterých došlo k neaditivní změně schématu, například k přejmenování nebo odstranění sloupce či ke změně datového typu. Rozdělte dotaz na rozsahy před a po změně schématu.
  • Last updated on