CREATE TABLE ... FLOW (kanály)

Important

Tato funkce je v beta verzi.

CREATE TABLE ... FLOW Pomocí příkazu vytvořte spravovanou tabulku v kanálu napsanou jedním nebo více toky.

Syntax

CREATE TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ flow_clause ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
        [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...] )

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

flow_clause
  FLOW [ IMPORT CHECKPOINT checkpoint_path ] INSERT [ONCE] BY NAME query

Pokud chcete do jedné spravované tabulky přidat více zdrojů, deklarujte několik toků, které ho cílí, pomocí příkazu CREATE FLOW (kanály):

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parameters

  • table_name

    Název spravované tabulky, která se má vytvořit. Pokud název není kvalifikovaný, vytvoří se tabulka v cílovém schématu kanálu. Název nesmí již patřit do tabulky streamování.

  • specifikace_tabulek

    Volitelně definuje sloupce, jejich typy, vlastnosti a popisy. Pokud ho vynecháte, schéma se odvodí z dotazu toku.

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

    Přidává očekávání kvality dat do spravované tabulky. Tato očekávání kvality dat lze sledovat v čase a přistupovat k nim prostřednictvím logu událostí pipeline. 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 může být složen z literálů, identifikátorů sloupců v tabulce a deterministických vestavěných SQL funkcí nebo operátorů kromě:

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

  • DĚLENÉ PODLE (sloupec [, ...])

    Volitelně rozdělí tabulku podle podmnožina sloupců.

  • CLUSTER BY klauzule

    Volitelně můžete v tabulce povolit shlukování kapalin . Nelze kombinovat PARTITIONED BY a CLUSTER BY.

  • Cesta k umístění

    Volitelné umístění úložiště pro data tabulky.

  • KOMENTÁŘ table_comment

    Literál STRING popisující tabulku.

  • TBLPROPERTIES – klauzule

    Volitelně nastaví jednu nebo více uživatelsky definovaných vlastností tabulky.

  • Klauzule WITH ROW FILTER

    Přidá do tabulky funkci filtru řádků. Budoucí dotazy na danou tabulku obdrží podmnožinu řádků, pro které se funkce vyhodnotí TRUE.

  • FLOW INSERT [JEDNOU] DOTAZ BY NAME

    Definuje tok připojení, který vloží výsledek query do tabulky a porovnává sloupce výsledků se sloupci tabulky podle názvu. query může odkazovat na dávkové nebo streamované zdroje. ONCE spustí tok jednou (například pro backfill) místo každé aktualizace. Každý pojmenovaný tok zpracovává svůj vstup přesně jednou pro aktualizaci kanálu, která je stejná jako FLOW INSERT BY NAME u tabulky streamování.

  • KONTROLNÍ checkpoint_path IMPORTU

    Literál STRING s cestou k existujícímu kontrolnímu bodu Structured Streaming, který lze importovat do toku, takže migrovaný stream pokračuje od svého posledního zavázaného offsetu místo opětovného zpracování zdroje. Import checkpointu je v beta fázi. Viz kontrolní bod Migrace strukturovaného streamování.

Migrujte kontrolní bod strukturovaného streamování

Important

Import checkpointu je v beta fázi.

Použijte IMPORT CHECKPOINT k migraci existujícího Structured Streaming workloadu do pipeline bez nutnosti znovu zpracovat zdrojový kód. Nastavte checkpoint_pathcheckpointLocation na váš dotaz Structured Streaming, který používáte, což může být cloudové úložiště, svazek Unity Catalog nebo cesta DBFS. Při první aktualizaci pipeline flow tento checkpoint klonuje do spravovaného úložiště pipeline. Tok pak pokračuje od posledního zavázaného posunu se svým stavem (například agregacemi, deduplikačními klíči a vodoznaky) neporušeným. Následné aktualizace pipeline používají klonovaný kontrolní bod toku; původní checkpoint není upravován.

Warning

Zastavte původní dotaz na strukturované streamování před spuštěním pipeline. Původní dotaz Structured Streaming lze po importu znovu použít, ale musíte spravovat jeho stav kontrolního bodu a zajistit, aby pipeline a dotaz nezapisovaly do stejné tabulky současně, což může vést k duplicitním datům.

Znovu vytvořte dotaz Structured Streaming jako tok pipeline, který pokračuje v zápisu do tabulky, do které dotaz zapsal, a importujte jeho kontrolní bod:

-- Use the existing table your Structured Streaming query writes to; CREATE TABLE adopts it.
CREATE TABLE streaming_table
-- Continue from the imported checkpoint instead of reprocessing the source.
FLOW IMPORT CHECKPOINT '/Volumes/my_catalog/my_schema/checkpoints/my_stream'
INSERT BY NAME
  -- The same source your original query read from.
  SELECT * FROM STREAM(source_table);

Kontrolní bod se importuje až při první aktualizaci potrubí. Pozdější aktualizace ignorují IMPORT CHECKPOINT. Úplné obnovení kontrolního bodu neimportuje znovu. Začíná od nového, prázdného kontrolního bodu a znovu zpracovává zdroj.

Omezení

  • Spravované tabulky nepodporují toky změn CDC. AUTO CDC INTO(SQL) nebo apply_changes / apply_changes_from_snapshot (Python) vůči spravované tabulce selže s MANAGED_TABLE_DOES_NOT_SUPPORT_CDCchybou . Pro cíle CDC použijte (CREATE STREAMING TABLEpipelines) s tokem definovaným v CREATE FLOW (pipelines).
  • Spravované tabulky nepodporují FLOW ... REPLACE WHERE. Podporuje se jenom FLOW INSERT BY NAME. Pro přepočítání a přepsání cílové podmnožiny tabulky vytvořte streamingovou tabulku a definujte REPLACE WHERE tok pomocí CREATE FLOW (pipelines). Viz Dávkové zpracování pomocí toků REPLACE WHERE.
  • Spravované tabulky se podporují jenom v kanálech s katalogem Unity. Metastor Hive se nepodporuje.
  • Název existující tabulky streamování pro spravovanou tabulku nelze znovu použít. Nejprve odstraňte streamovací tabulku nebo příkaz selže s CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLE.
  • IMPORT CHECKPOINTje podporována pouze na .FLOW INSERT BY NAME

Příklady

-- Create a managed table populated by an inline append flow from a streaming table
CREATE TABLE output
FLOW INSERT BY NAME SELECT * FROM STREAM(samples.tpch.orders);

-- Create a managed table that ingests files with schema inference and evolution
CREATE TABLE raw_data
FLOW INSERT BY NAME
  SELECT * FROM STREAM read_files('abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/base/path');

-- Create a partitioned managed table from a streaming source
CREATE TABLE events
PARTITIONED BY (bucket)
FLOW INSERT BY NAME
  SELECT id, bucket FROM STREAM read_files('abfss://my_path', format => 'json');

-- Create a managed table with liquid clustering
CREATE TABLE orders_clustered
CLUSTER BY (order_date, customer_id)
FLOW INSERT BY NAME
  SELECT
    o_orderkey   AS order_id,
    o_custkey    AS customer_id,
    o_orderdate  AS order_date,
    o_totalprice AS total_price
  FROM STREAM(samples.tpch.orders);

-- Create a managed table with a data quality expectation that drops violating rows
CREATE TABLE valid_events
  (CONSTRAINT positive_id EXPECT (id > 0) ON VIOLATION DROP ROW)
FLOW INSERT BY NAME
  SELECT id FROM STREAM read_files('s3://bucket/path', format => 'json');

Dodatečné zdroje

  • Poslední aktualizace v