Používejte samostatné streamovací tabulky

Samostatná streamovací tabulka je tabulka zaregistrovaná v katalogu Unity s dodatečnou podporou streamování nebo přírůstkového zpracování dat definovaných mimo kanál Lakeflow. Potrubí se automaticky vytvoří pro každou streamovací tabulku. Streamované tabulky můžete použít k přírůstkové načítání dat ze systému Kafka a cloudového úložiště objektů.

Můžete vytvářet a aktualizovat samostatné streamované tabulky ze služby Databricks SQL Warehouse nebo z poznámkového bloku spuštěného na obecném výpočetním prostředí bez serveru. Podrobnosti o rozdílech mezi dvěma možnostmi výpočetních prostředků najdete v tématu Požadavky pro samostatné kanály.

Pokud chcete vytvářet a aktualizovat samostatné streamovací tabulky pomocí Python z poznámkového bloku, přečtěte si téma Použití Python se samostatnými kanály.

Poznámka:

Informace o tom, jak používat tabulky Delta Lake jako zdroje a cíle datových proudů, najdete v článku Streamované čtení a zápis tabulek Delta Lake.

Požadavky

Informace o možnostech výpočetních prostředků, oprávněních a dalších požadavcích na vytváření, aktualizaci a dotazování samostatných streamovaných tabulek najdete v tématu Požadavky pro samostatné kanály.

Vytvořte streamované tabulky

Streamovací tabulka je definována dotazem SQL v Databricks SQL. Při vytváření streamované tabulky se data ve zdrojových tabulkách aktuálně používají k sestavení streamované tabulky. Potom aktualizujete tabulku, obvykle podle plánu, aby se všechna přidaná data ve zdrojových tabulkách připojila k streamované tabulce.

Když vytvoříte streamovací tabulku, považujete se za vlastníka tabulky.

Pokud chcete vytvořit streamovací tabulku z existující tabulky, použijte příkazCREATE STREAMING TABLE, jak je znázorněno v následujícím příkladu:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT product, price FROM STREAM raw_data;

V tomto případě se streamovací tabulka sales vytvoří z konkrétních sloupců raw_data tabulky s plánem, který se má aktualizovat každou hodinu. Použitý dotaz musí být streamovací dotaz. Pomocí klíčového STREAM slova můžete ke čtení ze zdroje použít sémantiku streamování.

Výpočetní prostředky používané k aktualizaci

Když pomocí příkazu CREATE OR REFRESH STREAMING TABLE vytvoříte streamovací tabulku, počáteční aktualizace dat a populace začnou okamžitě. Tyto operace nespotřebovávají výpočetní prostředky služby Databricks SQL Warehouse. Místo toho streamované tabulky spoléhají na kanály bez serveru pro vytvoření i aktualizaci. Pro každou streamovací tabulku se automaticky vytvoří a spravuje vyhrazené bezserverové potrubí systémem.

Načtení souborů pomocí Auto Loaderu

Pokud chcete vytvořit streamovací tabulku ze souborů v objemu, použijte Auto Loader. Pro většinu úloh na příjem dat z cloudového objektového úložiště použijte Auto Loader. Automatické nástroje pro nahrávání a datové kanály jsou navrženy tak, aby postupně a idempotentně načítaly stále rostoucí množství dat, jakmile dorazí do cloudového úložiště.

Pokud chcete v Databricks SQL použít automatický zavaděč, použijte funkci read_files. Následující příklady ukazují použití Auto Loaderu ke čtení množiny souborů JSON do streamovací tabulky:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT * FROM STREAM read_files(
    "/Volumes/my_catalog/my_schema/my_volume/path/to/data",
    format => "json"
  );

Ke čtení dat z cloudového úložiště můžete použít, také Auto Loader:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
    'abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json',
    format => "json"
  );

Další informace o automatickém zavaděči najdete v tématu Co je automatický zavaděč?. Další informace o používání Auto Loaderu v SQL najdete včetně příkladů v kapitole Načtení dat z úložiště objektů.

Streamovaný příjem z jiných zdrojů

Příklad příjmu dat z jiných zdrojů, včetně Kafka, najdete v tématu Načtení dat v kanálech.

Použít zachytávání změnových dat (CDC) pomocí automatických toků CDC

K zpracování záznamů o zachycení datových změn (CDC) ze zdroje do streamovací tabulky použijte klauzuli FLOW AUTO CDC. 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ů. Viz Zachytávání změn dat a snímky.

AUTO CDC zjednodušuje CDC tím, že automaticky zpracovává záznamy mimo objednávku. Zadáte klíče k identifikaci záznamů, sekvenčního sloupce pro řazení a zda se mají ukládat výsledky jako scD typu 1 (přímé aktualizace) nebo SCD typu 2 (sledování historie).

Následující příklad vytvoří streamovací tabulku, která použije změny CDC pomocí SCD typu 1:

CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  SEQUENCE BY sequenceNum
  STORED AS SCD TYPE 1;

Následující příklad používá SCD typu 2 k zachování historie změn:

CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  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;

Úplné podrobnosti o možnostech a chování automatického CDC najdete v tématu API pro automatické CDC: Zjednodušení zachycování změn dat pomocí kanálů. Úplný odkaz na syntaxi najdete v tématu CREATE STREAMING TABLE.

Použijte selektivní dávkové nahrazení v tocích REPLACE WHERE

Důležité

Tato funkce je v beta verzi. Streamovaná tabulka musí používat kanál Pipelines Preview.

FLOW REPLACE WHERE Pomocí klauzule můžete překompilovat a přepsat cílovou podmnožinu streamované tabulky bez opětovného zpracování celé historie tabulek. REPLACE WHERE toky jsou velmi vhodné pro přírůstkové dávkové zpracování joinů a agregací, opožděně doručených dat, opakované zpracování upstreamových dat, evoluci schématu a zpětné doplnění dat.

Úplné podrobnosti o REPLACE WHERE tocích, včetně požadavků, přepsání predikátu a přírůstkové aktualizace, najdete v tématu NAHRAZENÍ WHERE toků pro samostatné tabulky streamování.

Přijímat pouze nová data

Ve výchozím nastavení read_files funkce čte všechna existující data ve zdrojové složce během vytváření tabulky a pak zpracovává nově přicházející záznamy s každou aktualizací.

Chcete-li zabránit ingestování dat, která již existují ve zdrojové složce v době vytváření tabulky, nastavte možnost includeExistingFiles na false. To znamená, že se zpracovávají pouze data, která do složky přijdou po vytvoření tabulky. Například:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
    '/path/to/files',
    includeExistingFiles => false
  );

Nastavení běhového kanálu

Streamované tabulky vytvořené pomocí SQL warehouses se automaticky aktualizují pomocí pipeline. Pipeliny používají modul runtime v kanálu current ve výchozím nastavení. Informace o procesu vydávání najdete v poznámkách k verzi kanálů Lakeflow a v popisu procesu upgradu vydání.

Databricks doporučuje používat current kanál pro produkční úlohy. Nové funkce jsou nejprve vydány na preview kanálu. Můžete nastavit kanál pipeline na kanál preview a otestovat nové funkce zadáním preview jako vlastnosti tabulky pomocí příkazu CREATE OR REFRESH STREAMING TABLE. Chcete-li aktualizovat kanál pro existující streamovací tabulku, musíte spustit CREATE OR REFRESH STREAMING TABLE s aktualizovanou TBLPROPERTIES.

Následující příklad kódu ukazuje, jak nastavit kanál na náhled:

CREATE OR REFRESH STREAMING TABLE sales
  TBLPROPERTIES ('pipelines.channel' = 'preview')
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM raw_data;

Skrytí citlivých dat

Streamované tabulky můžete použít ke skrytí citlivých dat od uživatelů, kteří k tabulce přistupují. Jedním z přístupů je definovat dotaz tak, aby zcela vyloučil citlivé sloupce nebo řádky. Alternativně můžete použít masky sloupců nebo filtry řádků na základě oprávnění uživatele, který se dotazuje. Můžete například skrýt tax_id sloupec pro uživatele, kteří nejsou ve skupině HumanResourcesDept. Provedete to tak, že během vytváření tabulky streamování použijete syntaxi ROW FILTER a MASK. Další informace najdete v tématu Filtry řádků a masky sloupců.

Aktualizace streamované tabulky

Streamované tabulky automaticky vytvářejí a používají bezserverové kanály ke zpracování operací aktualizace. Obnova je řízena potrubím a aktualizace je monitorována skladem Databricks SQL, který se používá pro vytváření streamovací tabulky. Streamované tabulky je možné aktualizovat pomocí kanálu, který běží podle plánu.

I když máte naplánovanou aktualizaci, můžete kdykoli zavolat ruční aktualizaci. Aktualizace se zpracovávají stejným procesem, který byl automaticky vytvořen společně se streamovací tabulkou.

Aktualizace streamované tabulky:

REFRESH STREAMING TABLE sales;

Stav nejnovější aktualizace můžete zkontrolovat pomocí DESCRIBE TABLE funkce EXTENDED.

Poznámka:

Možná budete muset aktualizovat svou streamovací tabulku před použitím dotazů na cestování časem .

Informace o plánování aktualizace najdete v tématu Plánování aktualizací. Plánované aktualizace můžou obsahovat oznámení o aktualizacích a můžete nastavit režim výkonu aktualizace.

Jak funguje aktualizace

Aktualizace tabulky streamování vyhodnocuje pouze nové řádky, které přišly po poslední aktualizaci, a připojí pouze nová data.

Každá aktualizace používá k zpracování těchto nových dat aktuální definici tabulky streamování. Úprava definice streamované tabulky automaticky nepřepočítává existující data. Pokud je úprava nekompatibilní s existujícími daty (například změnou datového typu), další aktualizace selže s chybou.

Následující příklady vysvětlují, jak změny definice streamované tabulky ovlivňují chování aktualizace:

  • Odebrání filtru nezpracovává dříve filtrované řádky.
  • Změna projekcí sloupců nemá vliv na způsob zpracování existujících dat.
  • Spojení se statickými snímky používají stav snímku v době počátečního zpracování. Zpožděná příchozí data, která by odpovídala aktualizovanému snímku, se ignorují. To může vést k opomenutí faktů, pokud jsou dimenze zpožděné.
  • Úprava funkce CAST u existujícího sloupce způsobí chybu.

Pokud se vaše data změní způsobem, který nelze v existující streamovací tabulce podporovat, můžete provést úplnou aktualizaci.

Úplné obnovení streamovací tabulky

Úplné aktualizace znovu zpracovávají všechna data dostupná ve zdroji s nejnovější definicí. Nedoporučuje se volat úplné aktualizace zdrojů, které neuchovávají celou historii dat nebo mají krátkou dobu uchovávání, například Kafka, protože úplná aktualizace zkracuje stávající data. Pokud už data nejsou ve zdroji dostupná, možná nebudete moct obnovit stará data.

Například:

REFRESH STREAMING TABLE sales FULL;

Plánování a monitorování aktualizací

Streamovací tabulku můžete aktualizovat automaticky podle plánu nebo při změnách nadřazených dat a můžete nakonfigurovat časové limity aktualizace, oznámení a režimy výkonu. Viz Plán aktualizací.

Řízení přístupu k streamovaným tabulkám

Streamované tabulky podporují bohaté řízení přístupu, které podporují sdílení dat, a zároveň se vyhýbají potenciálně privátním datům. Vlastník tabulky streamování nebo uživatel s MANAGE oprávněním může udělit SELECT oprávnění jiným uživatelům. Uživatelé s SELECT přístupem k tabulce streamování nevyžadují SELECT přístup k tabulkám, na které odkazuje tabulka streamování. Toto řízení přístupu umožňuje sdílení dat při řízení přístupu k podkladovým datům.

Můžete také upravit vlastníka streamované tabulky.

Udělte oprávnění streamovací tabulce

Pokud chcete udělit přístup ke streamované tabulce, použijte tentoGRANT příkaz:

GRANT <privilege_type> ON <st_name> TO <principal>;

Může být privilege_type.

  • SELECT – uživatel může SELECT streamovací tabulku.
  • REFRESH – uživatel může REFRESH streamovací tabulku. Aktualizace se spouští pomocí oprávnění vlastníka.

Následující příklad vytvoří streamovací tabulku a udělí uživatelům oprávnění k výběru a aktualizaci:

CREATE OR REFRESH STREAMING TABLE st_name AS SELECT * FROM source_table;

-- Grant read-only access:
GRANT SELECT ON st_name TO read_only_user;

-- Grant read and refresh access:
GRANT SELECT ON st_name TO refresh_user;
GRANT REFRESH ON st_name TO refresh_user;

Další informace o udělení oprávnění k zabezpečitelným objektům katalogu Unity najdete v referenčních informacích k oprávněním katalogu Unity.

Odebrat oprávnění ze streamovací tabulky

Pokud chcete odvolat přístup z tabulky streamování, použijte tentoREVOKE příkaz:

REVOKE privilege_type ON <st_name> FROM principal;

Když SELECT dojde k odvolání oprávnění ke zdrojové tabulce od vlastníka streamované tabulky nebo jakéhokoliv jiného uživatele, kterému byla udělena MANAGE nebo SELECT oprávnění na streamovanou tabulku, nebo dojde k vyřazení zdrojové tabulky, vlastník streamované tabulky nebo uživatel s uděleným přístupem může stále dotazovat streamovanou tabulku. Dochází však k následujícímu chování:

  • Vlastník streamovací tabulky nebo jiní, kteří ztratili přístup ke streamovací tabulce, už nemohou REFRESH tuto streamovací tabulku, a streamovací tabulka časem zastarává.
  • Pokud je plán automatizovaný, další naplánovaný plán REFRESH selže nebo se nespustí.

Následující příklad odvolá SELECT oprávnění z read_only_user:

REVOKE SELECT ON st_name FROM read_only_user;

Změna vlastníka streamované tabulky

Uživatel s oprávněními MANAGE pro samostatnou tabulku streamování může prostřednictvím Průzkumníka katalogu nastavit nového vlastníka. Novým vlastníkem může být buď samotná osoba, nebo hlavní služba, na které má roli Uživatel Hlavní Služby.

  1. V pracovním prostoru Azure Databricks klikněte na ikonu Data. Otevřete Katalog pomocí Průzkumníka katalogu.

  2. Vyberte tabulku streamování, kterou chcete aktualizovat.

  3. Na pravém bočním panelu, v části O této streamovací tabulce, najděte vlastníka a klikněte na ikonu tužky pro úpravy.

    Poznámka:

    Pokud se zobrazí zpráva, která uvádí, že máte aktualizovat vlastníka změnou uživatele Run as v nastavení pipeline, pak je streamovací tabulka definována v pipeline Lakeflow, nikoli jako samostatná tabulka. Zpráva obsahuje odkaz na nastavení pipeline, kde můžete změnit Spustit jako uživatele.

  4. Vyberte nového vlastníka streamovací tabulky.

    Vlastníci mají automaticky oprávnění MANAGE a SELECT k streamovacím tabulkám, které vlastní. Pokud nastavíte služební účet jako vlastníka tabulky streamování, kterou vlastníte, a explicitně nemáte SELECT nebo MANAGE oprávnění k této tabulce streamování, tato změna by způsobila, že ztratíte veškerý přístup k tabulce. V takovém případě se zobrazí výzva k explicitní zadání těchto oprávnění.

    Vyberte jak oprávnění Udělit SPRÁVU, tak i Udělit SELECT práva, aby je bylo možné použít při Uložení.

  5. Kliknutím na Uložit změníte vlastníka.

Vlastník tabulky streamování je aktualizován. Všechny budoucí aktualizace se spouštějí pomocí identity nového vlastníka.

Když vlastník ztratí oprávnění ke zdrojovým tabulkám

Pokud změníte vlastníka a nový vlastník nemá přístup ke zdrojovým tabulkám (nebo SELECT oprávnění jsou u podkladových zdrojových tabulek odvolána), můžou se uživatelé stále dotazovat na streamovanou tabulku. Mějte však na paměti následující:

  • Nemůžou REFRESH tabulku streamování.
  • Další plánovaná aktualizace tabulky streamování selže.

Ztráta přístupu ke zdrojovým datům brání aktualizacím, ale nezneplatní okamžitě možnost čtení stávající streamované tabulky.

trvalé odstranění záznamů z tabulky streamování

Důležité

Podpora příkazu REORG s tabulkami streamování je ve verzi Public Preview.

Poznámka:

  • Použití příkazu REORG se streamovací tabulkou vyžaduje Databricks Runtime 15.4 a vyšší.
  • Ačkoli můžete použít REORG s libovolnou streamovací tabulkou, je ho potřeba jen při odstraňování záznamů ze streamovací tabulky s povolenými vektory odstranění. Příkaz nemá žádný vliv při použití se streamovanou tabulkou bez povolených vektorů odstranění.

Pokud chcete fyzicky odstranit záznamy z podkladového úložiště pro streamovanou tabulku s povolenými vektory odstranění, jako je například dodržování gdpr, je potřeba provést další kroky, které zajistí, aby VACUUM operace běžela na datech streamované tabulky.

Fyzické odstranění záznamů z podkladového úložiště:

  1. Aktualizujte záznamy nebo odstraňte záznamy z tabulky streamování.
  2. Spusťte příkaz REORG pro tabulku streamování a zadejte parametr APPLY (PURGE). Příklad: REORG TABLE <streaming-table-name> APPLY (PURGE);.
  3. Počkejte, až uplyne doba uchovávání dat streamované tabulky. Výchozí doba uchovávání dat je sedm dní, ale je možné ji nakonfigurovat pomocí vlastnosti delta.deletedFileRetentionDuration tabulky. Viz Konfigurace uchovávání dat pro časové dotazy.
  4. REFRESH streamovací tabulka. Viz Aktualizace streamované tabulky. Během 24 hodin po REFRESH operaci se úkoly údržby potrubí, včetně VACUUM operace potřebné k trvalému odstranění záznamů, automaticky spustí.

Monitorování běhů pomocí historie dotazů

Na stránce historie dotazů můžete získat přístup k podrobnostem dotazu a profilům dotazů, které vám pomůžou identifikovat špatně výkonné dotazy a kritické body v kanálu používaném ke spouštění aktualizací streamovaných tabulek. Přehled o druhu informací dostupných v historii dotazů a profilech dotazů najdete v tématu Historie dotazů a profil dotazu.

Důležité

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 . Podívejte se na Správa náhledů Azure Databricks.

Všechny příkazy související se streamovanými tabulkami se zobrazí v historii dotazů. Pomocí rozevíracího filtru Příkaz můžete vybrat libovolný příkaz a zkontrolovat související dotazy. Za všemi CREATE příkazy následuje REFRESH příkaz, který se v rámci pipeline spouští asynchronně. Příkazy REFRESH obvykle zahrnují podrobné plány dotazů, které poskytují přehled o optimalizaci výkonu.

Pokud chcete získat přístup k REFRESH příkazům v uživatelském rozhraní historie dotazů, postupujte následovně:

  1. Klikněte na ikonu Historie. Na levém bočním panelu otevřete uživatelské rozhraní Historie dotazů .
  2. Z rozbalovacího filtru REFRESH zaškrtněte políčko .
  3. Kliknutím na název příkazu dotazu zobrazíte souhrnné podrobnosti, jako je doba trvání dotazu a agregované metriky.
  4. Kliknutím na Zobrazit profil dotazu otevřete profil dotazu. Podrobnosti o navigaci v profilu dotazu najdete v části Profil dotazu.
  5. Volitelně můžete pomocí odkazů v části Zdroj dotazů otevřít související dotaz nebo kanál.

K podrobnostem dotazu můžete přistupovat také pomocí odkazů v editoru SQL nebo z poznámkového bloku připojeného k SQL Warehouse.

Přístup ke streamovaným tabulkám z externích klientů

Pro přístup ke streamovaným tabulkám z externích klientů Delta Lake nebo Iceberg, kteří nepodporují otevřená rozhraní API, můžete použít režim kompatibility. Režim kompatibility vytvoří verzi streamovací tabulky jen pro čtení, ke které má přístup jakýkoli klient Delta Lake nebo Iceberg.

Dodatečné zdroje

  • Last updated on