CREATE STREAMING TABLE (potrubí)

Streamovací tabulka je tabulka s podporou streamování nebo přírůstkového zpracování dat. Streamované tabulky jsou podporovány zpracovávacími linkami. Při každé aktualizaci streamované tabulky se data přidaná do zdrojových tabulek připojí ke streamované tabulce. Streamované tabulky můžete aktualizovat ručně nebo podle plánu.

Další informace o provádění nebo plánování aktualizací najdete v tématu Spuštění aktualizace pipeline.

Syntaxe

CREATE [OR REFRESH] [PRIVATE] STREAMING TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ {flow_clause | AS query} ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ column_constraint ] [, ...]
    [ , table_constraint ] [...] )

   column_properties
      { NOT NULL | GENERATED ALWAYS AS ( expr ) | GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY [ ( [ START WITH start | INCREMENT BY step ] [ ...] ) ] | DEFAULT default_expression | COMMENT column_comment | column_constraint | MASK clause } [ ... ]

table_clauses
  { USING DELTA
    PARTITIONED BY (col [, ...]) |
    CLUSTER BY clause |
    LOCATION path |
    COMMENT view_comment |
    TBLPROPERTIES clause |
    WITH { ROW FILTER clause } } [ ... ]
   } [ ... ]

flow_clause
  FLOW { { INSERT [ONCE] BY NAME query } |
  { AUTO CDC auto_cdc_flow_spec } |
  { REPLACE WHERE predicate BY NAME query } }

Parametry

  • REFRESH

    Pokud je tato hodnota zadaná, vytvoří se tabulka nebo aktualizuje existující tabulku a její obsah.

  • SOUKROMÝ

    Vytvoří soukromou tabulku pro streamování.

    • Nejsou přidány do katalogu a jsou přístupné pouze v rámci určeného potrubí.
    • Můžou mít stejný název jako existující objekt v katalogu. Pokud má tabulka privátního streamování a objekt v katalogu stejný název, odkazy na název se přeloží na tabulku privátního streamování.
    • Tabulky pro privátní streamování jsou uchovávány po celou dobu životnosti pipeline, nejen při jedné aktualizaci.

    Tabulky privátního streamování byly dříve vytvořeny pomocí parametru TEMPORARY .

  • table_name

    Název nově vytvořené tabulky. Plně kvalifikovaný název tabulky musí být jedinečný.

  • specifikace_tabulek

    Tato volitelná klauzule definuje seznam sloupců, jejich typů, vlastností, popisů a omezení sloupců.

    • column_identifier

      Názvy sloupců musí být jedinečné a mapované na výstupní sloupce dotazu.

    • column_type

      Určuje datový typ sloupce. Streamované tabulky nepodporují všechny datové typy podporované službou Azure Databricks.

    • column_comment

      Volitelný STRING literál popisující sloupec. Tato možnost musí být zadána společně s parametrem column_type. Pokud typ sloupce není zadaný, komentář sloupce se přeskočí.

    • VYGENEROVÁNO VŽDY JAKO ( výraz )

      Při zadání této klauzule je hodnota tohoto sloupce určena zadaným expr.

      DEFAULT COLLATION tabulky musí být UTF8_BINARY.

      expr se můžou skládat z literálů, identifikátorů sloupců v tabulce a deterministických integrovaných funkcí nebo operátorů SQL s výjimkou:

      Také expr nesmí obsahovat žádný poddotaz.

    • VYGENEROVÁNO { ALWAYS | VE VÝCHOZÍM REŽIMU } AS IDENTITY [ ( [ START WITH start ] [ INCREMENT BY step ] ) ]

      Platí pro:zaškrtnuto ano Databricks SQL zaškrtnuto ano Databricks Runtime 10.4 LTS nebo vyšší

      Definuje sloupec s identitou. Při zápisu do tabulky, pokud nezadáte hodnoty pro sloupec identity, bude automaticky přiřazena jedinečná hodnota, která se statisticky zvyšuje (nebo snižuje, pokud je step záporné). Tato klauzule je podporována pouze pro tabulky Delta. Tuto klauzuli lze použít pouze pro sloupce s datovým typem BIGINT.

      Automaticky přiřazené hodnoty začínají start a zvyšují se o step. Přiřazené hodnoty jsou jedinečné, ale nejsou zaručené, že budou souvislé. Oba parametry jsou volitelné a výchozí hodnota je 1. step nemůže být 0.

      Pokud jsou automaticky přiřazené hodnoty nad rámec rozsahu typu sloupce identity, dotaz selže.

      Při použití ALWAYS nemůžete pro sloupec identity zadat vlastní hodnoty.

      Následující operace nejsou podporovány:

      • PARTITIONED BY identifikační sloupec
      • UPDATE identifikační sloupec

      Poznámka:

      Deklarace sloupce identity v tabulce zakáže souběžné transakce. Sloupce identity používejte jenom v případech, kdy nejsou vyžadovány souběžné zápisy do cílové tabulky.

    • DEFAULT výchozí_výraz

      Platí pro:zaškrtnuto ano Databricks SQL zaškrtnuto ano Databricks Runtime 11.3 LTS a vyšší verze

      Definuje hodnotu DEFAULT pro sloupec, který se používá pro INSERT, UPDATEa MERGE ... INSERT, pokud sloupec není zadaný.

      Pokud není zadán žádný výchozí DEFAULT NULL se použije pro sloupce s možnou hodnotou null.

      default_expression mohou se skládat z literálů a integrovaných funkcí nebo operátorů SQL s výjimkou:

      Také default_expression nesmí obsahovat žádný poddotaz.

      DEFAULTpodporuje se pro CSV, JSONPARQUET, a ORC zdroje.

    • column_constraint

      Přidá omezení informačního primárního klíče nebo informačního cizího klíče do sloupce v tabulce streamování.

    • Klauzule MASKA

      Přidá funkci masky sloupce pro anonymizaci citlivých dat.

      Viz filtry řádků a masky sloupců.

    • CONSTRAINT expectation_name EXPECT (expectation_expr) [ PŘI PORUŠENÍ { SELHAT UPDATE | SMAZAT ŘÁDEK } ]

      Přidá do tabulky streamování očekávání kvality dat. Tato očekávání kvality dat je možné sledovat v průběhu času a k nim přistupovat prostřednictvím protokolu událostí streamované tabulky. Očekávání FAIL UPDATE způsobí selhání zpracování při tvorbě a aktualizaci tabulky. Očekávání DROP ROW způsobí, že pokud se očekávání nesplní, celý řádek bude vynechán. Viz Spravujte kvalitu dat pomocí požadavků na datový potrubí.

      expectation_expr se můžou skládat z literálů, identifikátorů sloupců v tabulce a deterministických integrovaných funkcí nebo operátorů SQL s výjimkou:

      Také expr nesmí obsahovat žádný poddotaz.

  • omezení tabulky

    Při zadávání schématu můžete definovat primární a cizí klíče. Omezení jsou informativní a nevynucují se. Viz klauzuli CONSTRAINT v referenční dokumentaci jazyka SQL.

    Poznámka:

    Pokud chcete definovat omezení tabulek, vaše potrubí musí být aktivováno prostřednictvím katalogu Unity.

  • tabulka_podmínky

    Volitelně můžete zadat vlastnosti dělení, komentářů a uživatelem definovaných vlastností tabulky. Každou dílčí klauzuli lze zadat pouze jednou.

    • POUŽITÍ FUNKCE DELTA

      Určuje formát dat. Jedinou možností je DELTA.

      Tato klauzule je volitelná a výchozí hodnota je DELTA.

    • PARTICIONOVÁNO PODLE

      Volitelný seznam jednoho nebo více sloupců, které se mají použít k dělení v tabulce. Vzájemně se vylučuje s CLUSTER BY.

      Liquid clustering poskytuje flexibilní a optimalizované řešení pro shlukování. Zvažte použití CLUSTER BY místo PARTITIONED BY pro kanály.

    • CLUSTER BY

      Povolte tekuté shlukování v tabulce a definujte sloupce, které se mají použít jako klíče shlukování. Pomocí automatického liquid clusteringu s CLUSTER BY AUTO a Databricks inteligentně vybírá klíče seskupování pro optimalizaci výkonu dotazů. Vzájemně se vylučuje s PARTITIONED BY.

      Viz Použití metody 'liquid clustering' pro tabulky.

    • UMÍSTĚNÍ

      Volitelné umístění úložiště pro data tabulky. Pokud není nastavená, systém ve výchozím nastavení nastaví umístění úložiště kanálu.

    • KOMENTÁŘ

      Volitelný STRING literál, který popisuje tabulku.

    • TBLPROPERTIES

      Volitelný seznam vlastností tabulky pro tabulku.

    • S ROW FILTER

    Přidá do tabulky funkci filtru řádků. Budoucí dotazy na danou tabulku obdrží podmnožinu řádků, pro které se funkce vyhodnotí jako PRAVDA. To je užitečné pro důkladné řízení přístupu, protože umožňuje funkci zkoumat identitu a členství ve skupinách uživatele, který ji vyvolává, aby se rozhodlo, zda mají být určité řádky filtrovány.

    Vizklauzule .

    • TOK

      Volitelně definuje tok vložený s vytvořením tabulky. Tok je stavový dotaz, který aktualizuje obsah tabulky. Pokud FLOW není zadán, můžete místo toho použít AS query nebo definovat toky samostatně pomocí CREATE FLOW. Můžete zadat jeden z následujících typů toků:

      • INSERT JMÉNEM

        Vloží data do tabulky podle názvu sloupce. Pokud tato ONCE možnost není zadána, musí být dotaz streamovaným dotazem. Pomocí klíčového STREAM slova můžete ke čtení ze zdroje použít sémantiku streamování. Pokud čtení narazí na změnu nebo odstranění existujícího záznamu, vyvolá se chyba. Je nejbezpečnější číst ze statických nebo doplňovacích zdrojů.

        Poznámka:

        FLOW INSERT BY NAME je ekvivalentní použití AS query. Následující dva příkazy mají stejné chování:

        CREATE OR REFRESH STREAMING TABLE raw_data
AS SELECT * FROM STREAM read_files('abfss://my_path');

CREATE OR REFRESH STREAMING TABLE raw_data
FLOW INSERT BY NAME SELECT * FROM STREAM read_files('abfss://my_path');

      • JEDNOU

        Volitelně definuje tok jako jednorázový tok, například jako backfill. Po ONCE zadání není dotaz streamováním a tok se ve výchozím nastavení spustí jednou. Pokud se tabulka aktualizuje úplnou aktualizací, ONCE tok se znovu spustí a znovu vytvoří data. ONCE platí pouze pro INSERT BY NAME toky.

      • AUTO CDC

        Důležité

        K dispozici ve službě Databricks Runtime 17.3 a novějších a kanálech PREVIEW Pipelines.

        AUTO CDC Definuje tok, který zpracovává záznamy CDC (Change Data Capture) ze zdroje do tabulky. Používá AUTO CDC se, když zdrojová data obsahují sémantiku CDC. Viz rozhraní API AUTO CDC: Zjednodušte zachytávání změn dat pomocí pipelin.

      • NAHRADIT WHEREpredikát BY NAME – dotaz

        Důležité

        FLOW REPLACE WHERE je v beta verzi. Vyžaduje použití kanálu Pipelines Preview – nastavte pipelines.channel vlastnost tabulky na "PREVIEW"hodnotu .

        REPLACE WHERE Definuje tok, který rekomputuje a přepíše pouze odpovídající predicateřádky , takže všechny ostatní řádky nebudou nedotčené. Slouží REPLACE WHERE k postupnému dávkovému zpracování spojení a agregací, pozdním příchodům dat, vývoji schématu a backfillům. BY NAME je povinné. Viz Dávkové zpracování s toky REPLACEWHERE.

  • Dotaz AS

    Tato klauzule naplní tabulku pomocí dat z query. Tento dotaz musí být streamovací. Pomocí klíčového slova STREAM můžete ke čtení ze zdroje použít sémantiku streamování. Pokud čtení narazí na změnu nebo odstranění existujícího záznamu, vyvolá se chyba. Je nejbezpečnější číst ze statických nebo doplňovacích zdrojů. Pokud chcete ingestovat data, která mají potvrzení změn, můžete přidat skipChangeCommits možnost čtení pro zpracování chyb.

    Když zadáte query a table_specification dohromady, schéma tabulky zadané v table_specification musí obsahovat všechny sloupce vrácené query, jinak se zobrazí chyba. Všechny sloupce uvedené v table_specification, které nejsou vráceny query, vracejí null hodnoty při dotazu.

    Další informace o streamovaných datech najdete v tématu Transformace dat pomocí kanálů.

    • Možnosti čtení

      V dotazu můžete zadat možnosti čtení a nakonfigurovat způsob čtení dat ze zdroje. Můžete například určit skipChangeCommits , že chcete přeskočit všechny potvrzení změn ve zdrojových datech. Možnosti čtení se zadají jako mapa v WITH klauzuli dotazu. Například:

      SELECT * FROM STREAM source_table WITH (SKIPCHANGECOMMITS=TRUE, STARTINGVERSION=X)

      Tato =TRUE možnost je volitelná, takže můžete také zadat logickou možnost:

      SELECT * FROM STREAM source_table WITH (SKIPCHANGECOMMITS)

      Poznámka:

      Možnosti čtení jsou podporovány pouze pro Databricks Runtime 17.3 a vyšší.

      Následující možnosti čtení jsou podporované pro Delta, podrobnosti o jednotlivých možnostech najdete v tématu Čtení a zápisy streamovaných tabulek Delta Lake.

      • maxFilesPerTrigger
      • maxBytesPerTrigger
      • startingVersion
      • startingTimestamp
      • readChangeFeed
      • withEventTimeOrder
      • skipChangeCommits

Požadovaná oprávnění

Uživatel spuštěn jako pro potrubí musí mít následující oprávnění:

  • SELECT oprávnění k základním tabulkám, na které odkazuje streamující tabulka.
  • USE CATALOG privilegium v nadřazeném katalogu a USE SCHEMA privilegium v nadřazeném schématu.
  • CREATE MATERIALIZED VIEW oprávnění ke schématu pro streamovací tabulku.

Aby uživatel mohl aktualizovat pipeline, ve které je definována tabulka streamování, potřebují:

  • USE CATALOG privilegium v nadřazeném katalogu a USE SCHEMA privilegium v nadřazeném schématu.
  • Vlastnictví streamované tabulky nebo REFRESH oprávnění na streamované tabulce.
  • Vlastník tabulky streamování musí mít SELECT oprávnění k základním tabulkám, na které odkazuje tabulka streamování.

Aby uživatel mohl dotazovat výslednou streamovací tabulku, vyžaduje:

  • USE CATALOG privilegium v nadřazeném katalogu a USE SCHEMA privilegium v nadřazeném schématu.
  • SELECT oprávnění pro streamovací tabulku.

Omezení

  • Nejnovější data můžou získat jenom vlastníci tabulek, kteří můžou aktualizovat streamované tabulky.
  • ALTER TABLE příkazy jsou u streamovaných tabulek zakázány. Definice a vlastnosti tabulky by se měly změnit prostřednictvím příkazu CREATE OR REFRESH nebo ALTER STREAMING TABLE.
  • Vývoj schématu tabulky pomocí příkazů DML, jako je INSERT INTO, a MERGE se nepodporuje.
  • U streamovaných tabulek se nepodporují následující příkazy:
    • CREATE TABLE ... CLONE <streaming_table>
    • COPY INTO
    • ANALYZE TABLE
    • RESTORE
    • TRUNCATE
    • GENERATE MANIFEST
    • [CREATE OR] REPLACE TABLE
  • Přejmenování tabulky nebo změna vlastníka se nepodporuje.

Examples

-- Define a streaming table from a volume of files:
CREATE OR REFRESH STREAMING TABLE customers_bronze
AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/customers/*", format => "csv")

-- Define a streaming table from a streaming source table:
CREATE OR REFRESH STREAMING TABLE customers_silver
AS SELECT * FROM STREAM(customers_bronze)

-- Use automatic liquid clustering to let Databricks choose the clustering columns:
CREATE OR REFRESH STREAMING TABLE customers_bronze_auto
CLUSTER BY AUTO
AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/customers/*", format => "csv")

-- Define a table with a row filter and column mask:
CREATE OR REFRESH STREAMING TABLE customers_silver (
  id int COMMENT 'This is the customer ID',
  name string,
  region string,
  ssn string MASK catalog.schema.ssn_mask_fn COMMENT 'SSN masked for privacy'
)
WITH ROW FILTER catalog.schema.us_filter_fn ON (region)
AS SELECT * FROM STREAM(customers_bronze)

-- Define a streaming table with an identity column:
CREATE OR REFRESH STREAMING TABLE customers_with_id (
  customer_id BIGINT GENERATED ALWAYS AS IDENTITY,
  name string,
  region string
)
AS SELECT name, region FROM STREAM(customers_bronze)

-- Define a streaming table that you can add flows into:
CREATE OR REFRESH STREAMING TABLE orders;

-- Define a streaming table with an inline append flow:
CREATE OR REFRESH STREAMING TABLE raw_data
FLOW INSERT BY NAME SELECT * FROM STREAM read_files('abfss://my_path');

-- Define a streaming table with an inline AUTO CDC flow:
CREATE OR REFRESH STREAMING TABLE target
FLOW AUTO CDC
FROM stream(cdc_data.users)
KEYS (userId)
SEQUENCE BY sequenceNum
STORED AS SCD TYPE 1;
  • Last updated on