Osvědčené postupy: Delta Lake

Článek
04/02/2024

Tento článek popisuje osvědčené postupy při používání Delta Lake.

Databricks doporučuje používat prediktivní optimalizaci. Viz Prediktivní optimalizace pro Delta Lake.

Při odstraňování a opětovném vytvoření tabulky ve stejném umístění byste měli vždy použít CREATE OR REPLACE TABLE příkaz. Viz Odstranění nebo nahrazení tabulky Delta.

Použití clusteringu liquid pro přeskočení optimalizovaných dat

Databricks doporučuje místo dělení na oddíly, pořadí vykreslování nebo jiných strategií organizace dat používat k optimalizaci rozložení dat pro přeskočení dat. Viz Použití liquid clusteringu pro tabulky Delta.

Komprimovat soubory

Prediktivní optimalizace se automaticky spouští OPTIMIZE a VACUUM příkazy ve spravovaných tabulkách Katalogu Unity. Viz Prediktivní optimalizace pro Delta Lake.

Databricks doporučuje často spouštět příkaz OPTIMIZE pro komprimování malých souborů.

Poznámka:

Tato operace neodebere staré soubory. Pokud je chcete odebrat, spusťte příkaz VACUUM .

Nahrazení obsahu nebo schématu tabulky

Někdy může být vhodné nahradit tabulku Delta. Příklad:

Zjistíte, že data v tabulce jsou nesprávná a chcete obsah nahradit.
Chcete přepsat celou tabulku a provést nekompatibilní změny schématu (například změnit typy sloupců).

I když můžete odstranit celý adresář tabulky Delta a vytvořit novou tabulku na stejné cestě, nedoporučuje se, protože:

Odstranění adresáře není efektivní. Odstranění adresáře obsahujícího velmi velké soubory může trvat hodiny nebo dokonce dny.
Ztratíte veškerý obsah v odstraněných souborech; Pokud odstraníte nesprávnou tabulku, je těžké ji obnovit.
Odstranění adresáře není atomické. Při odstraňování tabulky může souběžný dotaz, který čte tabulku, selhat nebo zobrazit částečnou tabulku.

Pokud nepotřebujete změnit schéma tabulky, můžete odstranit data z tabulky Delta a vložit nová data nebo tabulku aktualizovat tak, aby opravili nesprávné hodnoty.

Pokud chcete změnit schéma tabulky, můžete nahradit celou tabulku atomicky. Příklad:

Python

dataframe.write \
  .format("delta") \
  .mode("overwrite") \
  .option("overwriteSchema", "true") \
  .saveAsTable("<your-table>") # Managed table

dataframe.write \
  .format("delta") \
  .mode("overwrite") \
  .option("overwriteSchema", "true") \
  .option("path", "<your-table-path>") \
  .saveAsTable("<your-table>") # External table

SQL

REPLACE TABLE <your-table> USING DELTA AS SELECT ... -- Managed table
REPLACE TABLE <your-table> USING DELTA LOCATION "<your-table-path>" AS SELECT ... -- External table

Scala

dataframe.write
  .format("delta")
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable("<your-table>") // Managed table

dataframe.write
  .format("delta")
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .option("path", "<your-table-path>")
  .saveAsTable("<your-table>") // External table

Tento přístup má několik výhod:

Přepsání tabulky je mnohem rychlejší, protože nemusí rekurzivně vypisovat adresář ani odstraňovat žádné soubory.
Stará verze tabulky stále existuje. Pokud odstraníte nesprávnou tabulku, můžete snadno načíst stará data pomocí časového cestování. Přečtěte si: Práce s historií tabulky Delta Lake.
Je to atomická operace. Souběžné dotazy stále můžou číst tabulku, i když tabulku odstraňujete.
Z důvodu záruk transakcí Delta Lake ACID, pokud přepsání tabulky selže, bude tabulka ve svém předchozím stavu.

Kromě toho, pokud chcete odstranit staré soubory, abyste po přepsání tabulky ušetřili náklady na úložiště, můžete je odstranit pomocí funkce VAKUA . Je optimalizovaná pro odstranění souboru a obvykle je rychlejší než odstranění celého adresáře.

Ukládání do mezipaměti Sparku

Databricks nedoporučuje používat ukládání do mezipaměti Spark z následujících důvodů:

Přijdete o přeskočení dat, která můžou pocházet z dalších filtrů přidaných nad mezipaměť DataFrame.
Data, která se načte do mezipaměti, nemusí být aktualizována, pokud se k tabulce přistupuje pomocí jiného identifikátoru.

Rozdíly mezi Delta Lake a Parquet v Apache Sparku

Delta Lake zpracovává následující operace automaticky. Tyto operace byste nikdy neměli provádět ručně:

REFRESH TABLE: Tabulky Delta vždy vrací nejaktuálnější informace, takže po změnách není nutné volat REFRESH TABLE ručně.
Přidání a odebrání oddílů: Delta Lake automaticky sleduje sadu oddílů, které jsou přítomné v tabulce, a aktualizuje seznam při přidávání nebo odebírání dat. V důsledku toho není nutné spouštět ALTER TABLE [ADD|DROP] PARTITION ani MSCK.
Načtení jednoho oddílu: Čtení oddílů přímo není nutné. Například nemusíte spouštět spark.read.format("parquet").load("/data/date=2017-01-01"). Místo toho použijte WHERE klauzuli pro přeskočení dat, například spark.read.table("<table-name>").where("date = '2017-01-01'").
Neupravujte ručně datové soubory: Delta Lake používá transakční protokol k potvrzení změn v tabulce atomicky. Neupravujte, nepřidávejte ani neodstraňovat datové soubory Parquet v tabulce Delta, protože to může vést ke ztrátě dat nebo poškození tabulky.

Zvýšení výkonu pro sloučení Delta Lake

Pomocí následujících přístupů můžete zkrátit dobu potřebnou ke sloučení:

Zmenšete místo hledání shody: Ve výchozím nastavení merge operace prohledá celou tabulku Delta a vyhledá shody ve zdrojové tabulce. Jedním ze způsobů, jak urychlit hledání merge , je snížit prostor hledání přidáním známých omezení do podmínky shody. Předpokládejme například, že máte tabulku rozdělenou podle country a date chcete použít merge k aktualizaci informací za poslední den a konkrétní zemi. Když přidáte následující podmínku, dotaz se zrychlová, protože hledá shody pouze v příslušných oddílech:
```
events.date = current_date() AND events.country = 'USA'
```
Kromě toho tento dotaz také snižuje pravděpodobnost konfliktů s jinými souběžnými operacemi. Další podrobnosti najdete v tématu Úrovně izolace a konflikty zápisu v Azure Databricks .
Kompaktní soubory: Pokud jsou data uložená v mnoha malých souborech, může být čtení dat, která se mají vyhledat shody, pomalé. Kvůli zlepšení propustnosti čtení můžete komprimovat malé soubory do větších souborů. Podrobnosti najdete v tématu Kompaktní datové soubory s optimalizací v Delta Lake .
Řídit oddíly náhodného prohazování pro zápisy: Operace merge několikrát prohazuje data pro výpočet a zápis aktualizovaných dat. Počet úloh používaných k náhodnému prohazování je řízen konfigurací spark.sql.shuffle.partitionsrelace Sparku . Nastavení tohoto parametru nejen řídí paralelismus, ale také určuje počet výstupních souborů. Zvýšení hodnoty zvyšuje paralelismus, ale také generuje větší počet menších datových souborů.
Povolit optimalizované zápisy: U dělených tabulek merge může vzniknout mnohem větší počet malých souborů než počet oddílů náhodného prohazování. Důvodem je to, že každá úloha náhodného náhodného prohazu může zapisovat více souborů do více oddílů a může se stát kritickým bodem výkonu. Počet souborů můžete snížit povolením optimalizovaných zápisů. Viz Optimalizované zápisy pro Delta Lake v Azure Databricks.
Ladění velikostí souborů v tabulce: Azure Databricks dokáže automaticky zjistit, jestli tabulka Delta obsahuje časté merge operace, které přepisují soubory, a můžou se rozhodnout zmenšit velikost přepsaných souborů v očekávání dalších přepsání souborů v budoucnu. Podrobnosti najdete v části věnované ladění velikostí souborů.
Nízká sloučení shuffle: Nízká sloučení shuffle poskytuje optimalizovanou implementaci MERGE , která poskytuje lepší výkon pro většinu běžných úloh. Kromě toho zachovává stávající optimalizace rozložení dat, jako je řazení Z u neupravených dat.

Správa aktuálnosti dat

Na začátku každého dotazu se tabulky Delta automaticky aktualizují na nejnovější verzi tabulky. Tento proces lze pozorovat v poznámkových blocích, když příkaz hlásí stav: Updating the Delta table's state. Při spouštění historických analýz v tabulce ale nemusíte nutně potřebovat data v minutách, zejména u tabulek, ve kterých se streamovaná data často ingestují. V těchto případech je možné dotazy spustit na zastaralých snímcích tabulky Delta. Tento přístup může snížit latenci při získávání výsledků z dotazů.

Odolnost proti zastaralým datům můžete nakonfigurovat nastavením konfigurace spark.databricks.delta.stalenessLimit relace Sparku s hodnotou časového řetězce, například 1h nebo 15m (po dobu 1 hodiny nebo 15 minut). Tato konfigurace je specifická pro relaci a nemá vliv na ostatní klienty, kteří k tabulce přistupují. Pokud byl stav tabulky aktualizován v rámci limitu nečekanosti, vrátí dotaz na tabulku výsledky bez čekání na nejnovější aktualizaci tabulky. Toto nastavení nikdy nebrání aktualizaci tabulky a když se vrátí zastaralá data, procesy aktualizace na pozadí. Pokud je poslední aktualizace tabulky starší než limit neautnosti, dotaz nevrací výsledky, dokud se aktualizace stavu tabulky nedokončí.

Vylepšené kontrolní body pro dotazy s nízkou latencí

Delta Lake zapisuje kontrolní body jako agregovaný stav tabulky Delta s optimalizovanou frekvencí. Tyto kontrolní body slouží jako výchozí bod pro výpočet nejnovějšího stavu tabulky. Bez kontrolních bodů by Delta Lake musel číst velkou kolekci souborů JSON ("delta" souborů), které představují potvrzení do transakčního protokolu, aby bylo možné vypočítat stav tabulky. Kromě toho jsou statistiky na úrovni sloupců, které Delta Lake používá k provádění přeskočení dat, uloženy v kontrolním bodu.

Důležité

Kontrolní body Delta Lake se liší od kontrolních bodů strukturovaného streamování.

Statistiky na úrovni sloupců se ukládají jako struktura a JSON (kvůli zpětné kompatibilitě). Formát struktury umožňuje, aby Delta Lake četl mnohem rychleji, protože:

Delta Lake neprovádí nákladné analýzy JSON za účelem získání statistik na úrovni sloupců.
Možnosti vyřezávání sloupců Parquet výrazně snižují vstupně-výstupní operace potřebné ke čtení statistik pro sloupec.

Formát struktury umožňuje kolekci optimalizací, které snižují režii operací čtení Delta Lake z sekund na desítky milisekund, což výrazně snižuje latenci krátkých dotazů.

Správa statistik na úrovni sloupců v kontrolních bodech

Spravujete, jak se statistiky zapisují do kontrolních bodů pomocí vlastností delta.checkpoint.writeStatsAsJson tabulky a delta.checkpoint.writeStatsAsStruct. Pokud jsou falseobě vlastnosti tabulky , Delta Lake nemůže provádět přeskočení dat.

Služba Batch zapisuje statistiky zápisu ve formátu JSON i struktury. delta.checkpoint.writeStatsAsJson je true.
delta.checkpoint.writeStatsAsStruct je ve výchozím nastavení nedefinovaný.
Čtenáři používají sloupec struktury, pokud je k dispozici, a jinak se vrátí k použití sloupce JSON.

Důležité

Vylepšené kontrolní body nerušují kompatibilitu s opensourcovými čtečkami Delta Lake. Nastavení delta.checkpoint.writeStatsAsJson , které false může mít vliv na proprietární čtenáře Delta Lake. Pokud se chcete dozvědět více o dopadech na výkon, obraťte se na dodavatele.

Povolení rozšířených kontrolních bodů pro dotazy strukturovaného streamování

Pokud vaše úlohy strukturovaného streamování nemají požadavky na nízkou latenci (podminut latence), můžete povolit vylepšené kontrolní body spuštěním následujícího příkazu SQL:

ALTER TABLE [<table-name>|delta.`<path-to-table>`] SET TBLPROPERTIES
('delta.checkpoint.writeStatsAsStruct' = 'true')

Latenci zápisu kontrolního bodu můžete také zlepšit nastavením následujících vlastností tabulky:

ALTER TABLE [<table-name>|delta.`<path-to-table>`] SET TBLPROPERTIES
(
 'delta.checkpoint.writeStatsAsStruct' = 'true',
 'delta.checkpoint.writeStatsAsJson' = 'false'
)

Pokud v aplikaci není vynechání dat užitečné, můžete obě vlastnosti nastavit na false. Pak se neshromažďují ani nezapisují žádné statistiky. Databricks tuto konfiguraci nedoporučuje.