Použití změnového datového kanálu Delta Lake v Azure Databricks

Kanál změn dat umožňuje Službě Azure Databricks sledovat změny na úrovni řádků mezi verzemi tabulky Delta. Pokud je v tabulce Delta povolená, modul runtime zaznamenává změny událostí pro všechna data zapsaná do tabulky. To zahrnuje data řádků spolu s metadaty označujícími, jestli byl zadaný řádek vložený, odstraněný nebo aktualizovaný.

Přírůstkové zpracování dat o změnách

Databricks doporučuje používat strukturované streamování v kombinaci s protokolem změn k postupnému zpracovávání změn z tabulek Delta. Strukturované streamování pro Azure Databricks musíte použít k automatickému sledování verzí pro datový kanál změn vaší tabulky.

Pokud chcete číst datový kanál změn z tabulky, musíte ho v této tabulce povolit. Podívejte se na Povolení změn dat.

Nastavte možnost readChangeFeed na true, když konfigurujete datový proud proti tabulce pro čtení kanálu změnových dat, jak je znázorněno v následujícím příkladu syntaxe:

Python

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

Scala

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

Ve výchozím nastavení datový proud vrátí nejnovější snímek tabulky při prvním spuštění streamu jako INSERT a následné změny jako změnová data.

Potvrzení změn dat probíhá jako součást transakce Delta Lake a stává se dostupným ve stejnou dobu, kdy se nová data zapisují do tabulky.

Volitelně můžete zadat počáteční verzi. Viz Mám zadat počáteční verzi?.

Kanál změn dat také podporuje dávkové spouštění, které vyžaduje zadání počáteční verze. Viz Čtení změn v dávkových dotazech.

Možnosti, jako jsou limity rychlosti (maxFilesPerTrigger, maxBytesPerTrigger) a excludeRegex jsou také podporovány při čtení dat o změnách.

Omezení rychlosti může být atomické pro jiné verze než počáteční verze snímku. To znamená, že celá verze potvrzení bude omezená rychlostí nebo se vrátí celé potvrzení.

Mám zadat počáteční verzi?

Pokud chcete ignorovat změny, ke kterým došlo před konkrétní verzí, můžete volitelně zadat počáteční verzi. Verzi můžete zadat pomocí časového razítka nebo čísla ID verze zaznamenaného v transakčním protokolu Delta.

Při konfiguraci úloh strukturovaného streamování zahrnujících datový kanál změn je důležité pochopit, jak určení počáteční verze ovlivňuje zpracování.

Mnohé úlohy streamování, zejména nové pipeliny pro zpracování dat, těží z výchozího chování. Při výchozím chování se první dávka zpracuje, když stream nejdříve zaznamená všechny existující záznamy v tabulce jako operace INSERT v kanálu změn dat.

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í.

Příklad syntaxe níže ukazuje obnovení po selhání streamování, při kterém byl poškozen kontrolní bod. 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á podřízená tabulka zpracovala všechny změny až do verze 75 a včetně ní.
3. Historie verzí zdrojové tabulky je k dispozici pro verze 70 a vyšší.

Python

(spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")
)

Scala

spark.readStream
  .option("readChangeFeed", "true")
  .option("startingVersion", 76)
  .table("source_table")

V tomto příkladu musíte také zadat nové umístění kontrolního bodu.

Podívejte se na Mohutíme použít změnu kanálu dat k přehrání celé historie 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í.

Jako celé číslo specifikujete verzi a časové razítko jako řetězec ve formátu yyyy-MM-dd[ HH:mm:ss[.SSS]].

Počáteční a koncové verze jsou v dotazech inkluzivní. Pokud chcete přečíst změny z konkrétní počáteční verze na nejnovější verzi tabulky, zadejte pouze počáteční verzi.

Pokud zadáte verzi nižší nebo časové razítko starší, než je to, které zaznamenalo události změn – tedy bylo starší než při spuštění záznamu změn – zobrazí se chyba, že záznam změn nebyl aktivován.

Následující příklady syntaxe ukazují použití možností počáteční a koncové verze při dávkovém čtení.

SQL

-- version as ints or longs e.g. changes from version 0 to 10
SELECT * FROM table_changes('tableName', 0, 10)

-- timestamp as string formatted timestamps
SELECT * FROM table_changes('tableName', '2021-04-21 05:45:46', '2021-05-21 12:00:00')

-- providing only the startingVersion/timestamp
SELECT * FROM table_changes('tableName', 0)

-- database/schema names inside the string for table name, with backticks for escaping dots and special characters
SELECT * FROM table_changes('dbName.`dotted.tableName`', '2021-04-21 06:45:46' , '2021-05-21 12:00:00')

Python

# version as ints or longs
spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .option("endingVersion", 10) \
  .table("myDeltaTable")

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

# providing only the startingVersion/timestamp
spark.read \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 0) \
  .table("myDeltaTable")

Scala

// version as ints or longs
spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .option("endingVersion", 10)
  .table("myDeltaTable")

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

// providing only the startingVersion/timestamp
spark.read
  .option("readChangeFeed", "true")
  .option("startingVersion", 0)
  .table("myDeltaTable")

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

Jaké je schéma kanálu změn dat?

Při čtení z datového kanálu změn pro tabulku se použije schéma nejnovější verze tabulky.

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

(1)preimage je hodnota před aktualizací, postimage je hodnota po aktualizaci.

Povolení toku změn dat

Datový proud změn můžete číst jen u povolených tabulek. Možnost změn datového kanálu musíte povolit explicitně pomocí jedné 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)
  

• Existující tabulka: Nastavte vlastnost delta.enableChangeDataFeed = true tabulky v ALTER TABLE příkazu.

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

• Všechny nové tabulky:

  set spark.databricks.delta.properties.defaults.enableChangeDataFeed = true;
  

Změna úložiště dat

Povolení funkce změnového datového kanálu způsobí malé zvýšení nákladů na úložiště pro tabulku. Při spuštění dotazu se vygenerují záznamy změn dat a jsou obecně mnohem menší než celková velikost přepsaných souborů.

Azure Databricks zaznamenává změny dat pro UPDATE, DELETE a MERGE operace ve složce _change_data pod adresářem tabulky. Některé operace, jako jsou operace jen pro vložení a odstranění celého oddílu, negenerují data v _change_data adresáři, protože Azure Databricks dokáže efektivně vypočítat datový kanál změn přímo z transakčního protokolu.

Všechna čtení datových souborů ve _change_data složce by měla projít podporovanými rozhraními Delta Lake API.

Soubory ve _change_data složce se řídí zásadami uchovávání informací v tabulce. Údaje o změnách datového kanálu se odstraní při spuštění příkazu VACUUM.

Můžu pomocí kanálu změn dat přehrát celou historii tabulky?

Datový kanál změn není určen jako trvalý záznam všech změn v tabulce. Kanál změn dat zaznamenává pouze změny, ke kterým dojde po jeho povolení.

Změnový kanál a Delta Lake vám umožňují vždy zrekonstruovat úplný snímek zdrojové tabulky, což znamená, že můžete zahájit nové streamové čtení proti tabulce, která má povolený kanál změn, a zachytit aktuální verzi této tabulky a všechny změny, ke kterým dojde následně.

Záznamy v datovém kanálu změn musíte považovat za přechodné a přístupné pouze pro zadané okno uchovávání informací. Protokol transakcí Delta v pravidelných intervalech odebere verze tabulek a jejich odpovídající verze datového kanálu změn. Při odebrání verze z transakčního protokolu už nemůžete číst datový kanál změn pro danou verzi.

Pokud váš případ použití vyžaduje zachování trvalé historie všech změn v tabulce, měli byste k zápisu záznamů z datového kanálu změn do nové tabulky použít přírůstkovou logiku. Následující příklad kódu ukazuje použití trigger.AvailableNow, který využívá inkrementální zpracování strukturovaného streamování, ale zpracovává dostupná data jako dávkovou úlohu. Tuto úlohu můžete naplánovat asynchronně s hlavními kanály zpracování a vytvořit zálohu datového kanálu změn pro účely auditování nebo úplné přehrá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")

Změna omezení datového kanálu pro tabulky s povoleným mapováním sloupců

Pokud je u tabulky Delta povolené mapování sloupců, můžete v tabulce přetáhnout nebo přejmenovat sloupce bez přepsání datových souborů pro existující data. Při povoleném mapování sloupců má datový kanál změn omezení po provedení neaditivních změn schématu, jako je přejmenování nebo zrušení sloupce, změna datového typu nebo změny nulovatelnosti.

Ve službě Databricks Runtime 12.2 LTS a vyšší můžete provádět dávkové čtení změnového datového proudu pro tabulky s povoleným mapováním sloupců, u kterých došlo k neaditivním změnám schématu. Místo použití schématu nejnovější verze tabulky operace čtení používají schéma koncové verze tabulky zadané v dotazu. Dotazy stále selžou, pokud zadaný rozsah verzí zahrnuje změnu schématu, která není aditivní.