CREATE FLOW (kanály)

Pomocí příkazu CREATE FLOW vytvořte toky nebo backfilly pro tabulky v kanálu.

Note

CREATE FLOW Cílení na streamovací tabulku podporuje jak AUTO CDC ... INTO toky, tak REPLACE WHERE toky. Spravovaná tabulka vytvořená s nepodporuje CREATE TABLE ... FLOW zachycení změn dat: tok AUTO CDC ... INTO proti spravované tabulce selže s .MANAGED_TABLE_DOES_NOT_SUPPORT_CDC CREATE FLOW Do streamovacího stolu tomuto omezení nepodléhají. Viz CREATE TABLE ... FLOW (potrubí).

Syntaxe

CREATE FLOW flow_name [COMMENT comment] AS
{
  AUTO CDC [ONCE] INTO target_table create_auto_cdc_flow_spec |
  AUTO CDC [ONCE] INTO target_table create_auto_cdc_from_snapshot_spec |
  INSERT [ONCE] INTO target_table BY NAME [ replace_using_spec | REPLACE WHERE condition ] query
}

create_auto_cdc_from_snapshot_spec
  FROM SNAPSHOT ( snapshot_query )
  [ WITH VERSION ( version_query ) ]
  KEYS ( key [, ...] )
  [ STORED AS { SCD TYPE 1 | SCD TYPE 2 } ]
  [ TRACK HISTORY ON { col_list | * EXCEPT ( col_list ) } ]

replace_using_spec
  REPLACE USING ( column_name [, ...] ) SEQUENCE BY sequence_column

Parametry

  • flow_name

    Název toku, který chcete vytvořit.

  • KOMENTÁŘ

    Volitelný popis toku.

  • AUTO CDC DO

    Výrok AUTO CDC ... INTO, který definuje tok, s create_auto_cdc_flow_spec. Musíte buď zahrnout AUTO CDC ... INTO příkaz, nebo příkaz INSERT INTO . Používá AUTO CDC ... INTO se, když zdrojový dotaz používá sémantiku změn dat.

    Další informace najdete v tématu AUTO CDC INTO (pipelines).

  • AUTO CDC ... Z SNAPSHOT

    Tvrzení, AUTO CDC ... INTO které odvozuje změny porovnáním snapshotů místo čtení feed změn. Tento formulář použijte tehdy, když není zachycení změn dat ve zdroji povoleno a jsou dostupné pouze úplné snímky. Zdroj je specifikován ve dvou částech: povinné klauzuli FROM SNAPSHOT (snapshot_query) , která čte data snapshotu, a volitelné WITH VERSION (version_query) klauzule, která vybírá další verzi snapshotu k zpracování. Podívejte se, jak funguje FUNKCE AUTO CDC FROM SNAPSHOT.

    • Z SNAPSHOT (snapshot_query)

      Required. Dotaz, který čte data snímku pro verzi vybranou .WITH VERSION (...) Engine rozlišuje výsledek vůči dříve zavázanému snapshotu, aby odvodil vložení, aktualizace a mazání, a spojuje je do cíle pomocí KEYS identity řádku a STORED AS určuje, jak jsou změny uloženy.

      Zavolejte current_snapshot_version() uvnitř tohoto dotazu pro odkaz na verzi vybranou .WITH VERSION (...) Pokud WITH VERSION (...) není specifikováno, current_snapshot_version() není volatelný uvnitř FROM SNAPSHOT (...).

      Pokud WITH VERSION (...) je vynechán, engine čte zdroj přímo přes FROM SNAPSHOT (...), a dotaz na snapshot běží pouze během počátečního načítání, zatímco cíl nemá žádná zavázaná data ani zavázaný stav snapshotu. Při jakékoli pozdější aktualizaci, když cíl již obsahuje data nebo má zadaný stav snapshotu, tok selže s .AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION Pro zpracování snapshotů napříč více aktualizacemi použijte WITH VERSION (...).

    • S VERZÍ (version_query)

      Optional. Dotaz, který vybere další verzi snapshotu ke zpracování. Musí vrátit přesně jeden sloupec pořadatelného typu a buď 0 nebo 1 řádky. Když vrátí 1 řádek, hodnota musí být nenulová. Sloupec může být skalární hodnota, například , BIGINTnebo , STRUCT jehož pole jsou všechna uspořádatelná. Dotaz na verzi, který vrací více než jeden sloupec, více než jeden řádek nebo nulovou hodnotu, selže tok s .INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY

      Během jedné aktualizace pipeline engine opakuje následující kroky: vyhodnocuje dotaz na verzi; pokud dotaz vrátí 0 řádků, ukončí zpracování tohoto toku pro aktuální aktualizaci; pokud dotaz vrátí 1 řádek, engine tuto hodnotu vystaví přes current_snapshot_version(), vyhodnotí dotaz na snapshot, potvrtí výsledný snapshot a exponuje popsanou verzi přes last_snapshot_version(). Engine pak znovu vyhodnotí dotaz na verzi, aby vybral další verzi. Jedna aktualizace pipeline zpracovává verze v pořadí, dokud dotaz na verzi nevrátí žádné řádky.

      Každá verze vrácená po úspěšném commitu musí být větší než dříve commitovaná verze; ne-klesající verze selže aktualizaci s APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION. Datový typ verze musí zůstat nezměněn při všech snapshot commitech; změna datového typu selže při aktualizaci s AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Kompletní obnovení vymaže stav přetrvávající verze.

    • KLÍČE

      Required. Primární klíčové sloupce slouží k identifikaci řádků napříč snímky pro detekci změn.

    • STORED AS { SCD TYPE 1 | SCD TYPE 2 }

      Optional. Specifikuje, jak jsou změny uloženy v cílové tabulce. Výchozí hodnota je SCD TYPE 1.

    • TRACK HISTORY ON { col_list | * EXCEPT (col_list) }

      Optional. Platí pouze s .SCD TYPE 2 Specifikuje, které sloupce spustí nový řádek historie, když se změní. Buď poskytněte explicitní seznam sloupců, nebo * EXCEPT (col_list) sledujte každý sloupec kromě těch uvedených.

    Snapshot CDC nepodporuje WHERE ani SEQUENCE BY. Uspořádání křížových snímků je vyjádřeno pomocí WITH VERSION (...).

  • target_table

    Tabulka, která se má aktualizovat. Musí se jednat o streamovací tabulku.

  • INSERT DO

    Definuje dotaz tabulky, který se vloží do cílové tabulky. Pokud tato ONCE možnost není zadána, musí být dotaz streamovaným dotazem. 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ů. K ingestci dat, která mají commity změn, můžete použít Python a možnost skipChangeCommits pro zpracování chyb.

    INSERT INTO se vzájemně vylučuje s AUTO CDC ... INTO. Používá se AUTO CDC ... INTO , když zdrojová data obsahují funkci pro zachytávání dat změn (CDC). Použijte INSERT INTO, pokud zdroj ne.

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

  • NAHRAZENÍ POMOCÍ ( column_name [, ...] ) POŘADÍ PODLE sequence_column

    Important

    Tato funkce je v beta verzi. Vyžaduje Databricks runtime 18.2 a vyšší.

    Definuje tok jako REPLACE USING tok, který nahrazuje všechny řádky v cílové tabulce odpovídající specifikovaným klíčovým sloupcům a všechny ostatní řádky ponechává nedotčené. Použijte REPLACE USING , když je zdroj série částečných snímků podle sloupců. SEQUENCE BY Pořadí aktualizací tak, aby vyhrála nejvyšší sekvence klíče, i když aktualizace přicházejí mimo pořadí.

    Specifikujte alespoň jeden klíčový sloupec a přesně jeden SEQUENCE BY sloupec. Dotaz musí být streamovaný a BY NAME je vyžadován. REPLACE USING Nelze kombinovat s ONCE ani s AUTO CDC ... INTO.

    Pro více informací viz Částečná výměna snímků pomocí NAHRAZENÍ POUŽÍVAJÍCÍCH toků.

  • Podmínka REPLACE WHERE

    Definuje tok jako REPLACE WHERE tok, který přepočítá a přepíše cílovou podmnožinu cílové tabulky. Při každé aktualizaci jsou všechny řádky v cílové condition tabulce smazány, zdrojový dotaz se přepočítá pro stejný rozsah predikátů a výsledky jsou vloženy. Řady, které nesedí condition , zůstávají nedotčené. Nemusíte pridávat predikát do zdrojového dotazu; pipeline engine jej automaticky aplikuje při čtení ze zdroje.

    REPLACE WHERE používá dávkovou sémantiku, takže zdrojový dotaz nemusí být streamovaný dotaz. BY NAME je povinné. REPLACE WHERE Nelze kombinovat s ONCE, REPLACE USING, nebo AUTO CDC ... INTO.

    Pro více informací viz Dávkové zpracování s toky REPLACE WHERE a REPLACE WHERE toky pro samostatné streamovací tabulky.

  • JEDNOU

    Volitelně můžete tok definovat jako jednorázový tok, například jako backfill. Použití ONCE změní tok dvěma způsoby:

    • Zdroj query nebo create_auto_cdc_flow_spec není streamovací tabulka.
    • Proud je ve výchozím nastavení spuštěn jednou. Pokud je potrubí aktualizováno o úplnou obnovu, ONCE tok se spustí znovu, aby znovu vytvořil data.

    ONCE nelze použít s REPLACE USING, což vyžaduje streamovací zdroj.

Examples

-- EXAMPLE 1:
-- Create a streaming table, and add two flows that append data to it:
CREATE OR REFRESH STREAMING TABLE users;

-- first flow into target_table:
CREATE FLOW users_flow AS
INSERT INTO users BY NAME
SELECT * FROM stream(raw_data.users);

-- second flow into target_table:
CREATE FLOW backfill_users AS
INSERT ONCE INTO users BY NAME
SELECT * FROM user_backfill_table;

-- EXAMPLE 2:
-- Create a streaming table, and add a flow that applies CDC changes to it:
CREATE OR REFRESH STREAMING TABLE admins_cdc_target_table;

-- first flow into target_table:
CREATE FLOW admin_cdc_flow AS
AUTO CDC INTO admins_cdc_target_table
FROM stream(cdc_data.admins)
KEYS (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY sequenceNum
COLUMNS * EXCEPT (operation, sequenceNum)
STORED AS SCD TYPE 2;

-- EXAMPLE 3:
-- Create a streaming table, and add a REPLACE USING flow that keeps the latest
-- row for each payment_id from a stream of partial snapshots:
CREATE OR REFRESH STREAMING TABLE payments_latest;

CREATE FLOW payments_replace_flow AS
INSERT INTO payments_latest BY NAME
REPLACE USING (payment_id) SEQUENCE BY payment_date
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

-- EXAMPLE 4:
-- AUTO CDC FROM SNAPSHOT without WITH VERSION: a one-time initial load from a snapshot table.
-- To process later snapshots on each update, add WITH VERSION (see EXAMPLE 5).
CREATE STREAMING TABLE users (user_id INT, name STRING, email STRING);

CREATE FLOW users_snapshot_flow AS
AUTO CDC ONCE INTO users
FROM SNAPSHOT (SELECT * FROM catalog.schema.users_snapshot)
KEYS (user_id)
STORED AS SCD TYPE 1;

-- EXAMPLE 5:
-- AUTO CDC FROM SNAPSHOT with WITH VERSION: pick the next file, then read it as the snapshot:
CREATE STREAMING TABLE orders (order_id INT, product STRING, quantity INT, order_date DATE);

CREATE FLOW orders_cdc AS
AUTO CDC INTO orders
FROM SNAPSHOT (
  SELECT order_id, product, quantity, order_date
  FROM read_files('/Volumes/catalog/schema/landing/orders/', format => 'json')
  WHERE _metadata.file_path = (SELECT version.path FROM current_snapshot_version())
)
WITH VERSION (
  SELECT struct(modification_time, path) AS version
  FROM list_files('/Volumes/catalog/schema/landing/orders/')
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR struct(modification_time, path) > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY modification_time, path
  LIMIT 1
)
KEYS (order_id)
STORED AS SCD TYPE 2;

-- EXAMPLE 6:
-- One-time snapshot backfill plus a streaming CDC flow into the same target.
-- The backfill omits WITH VERSION, so it uses an implicit timestamp version. The
-- streaming flow's SEQUENCE BY column (event_ts) must be a TIMESTAMP so its type
-- matches that implicit version on the shared target.
CREATE STREAMING TABLE customers (
  customer_id INT, name STRING, email STRING, address STRING, event_ts TIMESTAMP
);

CREATE FLOW customers_snapshot_backfill AS
AUTO CDC ONCE INTO customers
FROM SNAPSHOT (SELECT * FROM catalog.schema.customers_snapshot)
KEYS (customer_id)
STORED AS SCD TYPE 1;

CREATE FLOW customers_cdc AS
AUTO CDC INTO customers
FROM STREAM(customers_cdc_events)
KEYS (customer_id)
SEQUENCE BY event_ts
STORED AS SCD TYPE 1;

-- EXAMPLE 7:
-- Create a streaming table, then add a REPLACE WHERE flow that recomputes and
-- overwrites a targeted window of the target table on each update:
CREATE STREAMING TABLE payments_latest;

CREATE FLOW payments_latest AS
INSERT INTO payments_latest BY NAME
REPLACE WHERE payment_date >= date_add(current_date(), -7)
SELECT payment_id, booking_id, status, payment_date
FROM samples.wanderbricks.payments;

Více o kombinaci jednorázového doplňování s probíhajícím CDC na stejném cíli najdete v článku Doplňování historických dat pomocí potrubí.

  • Poslední aktualizace v