CREATE FLOW (kanály)

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

Syntaxe

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

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).

  • 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ů.

  • 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);
  • Last updated on