CREATE FLOW (csővezetékek)

A CREATE FLOW utasítással folyamatokat vagy visszatöltéseket hozhat létre egy csővezeték tábláihoz.

Note

CREATE FLOW A streaming tábla célzása mind AUTO CDC ... INTO a folyamatokat, REPLACE WHERE mind a flow-okat támogatja. Egy kezelt tábla, amelyet a kezelt táblával CREATE TABLE ... FLOW hoztak létre, nem támogatja a változásadat-rögzítést: egy AUTO CDC ... INTO menedzselt táblával szembeni folyamat meghibásodik .MANAGED_TABLE_DOES_NOT_SUPPORT_CDC CREATE FLOW egy streaming táblára nem vonatkozik ez a korlátozás. Lásd...CREATE TABLE FLOW (csővezetékek).

Szemantika

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

Paraméterek

  • flow_name

    A létrehozandó folyamat neve.

  • MEGJEGYZÉS

    A folyamat opcionális leírása.

  • AUTOMATIKUS CDC-BE

    Egy AUTO CDC ... INTO utasítás, amely meghatározza a folyamatot egy create_auto_cdc_flow_spec. Tartalmaznia kell vagy egy AUTO CDC ... INTO utasítást, vagy egy INSERT INTO utasítást. Akkor használható AUTO CDC ... INTO , ha a forrás lekérdezés változásadat-szemantikát használ.

    További információ: AUTO CDC INTO (adattovábbítási csatornák).

  • AUTO CDC ... SNAPSHOTBÓL

    Egy AUTO CDC ... INTO olyan állítás, amely a változásokat a pillanatképek összehasonlításával vezeti le, nem pedig egy változási hírfolyamot. Használja ezt az űrlapot, ha a forrásban nincs engedélyezve a változás adatrögzítése, és csak teljes pillanatképek állnak rendelkezésre. A forrás két részből áll: egy kötelező FROM SNAPSHOT (snapshot_query) záradék, amely olvassa a snapshot adatokat, és egy opcionális WITH VERSION (version_query) záradék, amely kiválasztja a következő snapshot verziót feldolgozni. Tekintse meg , hogyan működik az AUTO CDC FROM SNAPSHOT.

    • SNAPSHOT (snapshot_query) FORRÁSÁBÓL

      Required. Egy lekérdezés, amely a kiválasztott verzió WITH VERSION (...)pillanatkép adatait olvassa . A motor differálja az eredményt a korábban elkötelezett pillanatképtől, hogy beillesztéseket, frissítéseket és törléseket lehessen vezetni, majd sorazonosításhoz és STORED AS a változtatások tárolásának meghatározásához használja őket a célponttalKEYS.

      Hívjuk meg a current_snapshot_version() lekérdezést ebben a lekérdezésben, hogy hivatkozzon a kiválasztott verzióra WITH VERSION (...). Ha WITH VERSION (...) nincs megadva, current_snapshot_version() nem hívható be -n belül FROM SNAPSHOT (...).

      Ha WITH VERSION (...) kihagyják, a motor közvetlenül a forrást olvasza FROM SNAPSHOT (...), és a snapshot lekérdezés csak az első betöltés alatt fut, miközben a célpontnak nincs kötelezett adata és nincs elkötelezett snapshot állapota. Későbbi frissítésnél, amikor a célpont már tartalmaz adatot vagy elkötelezett pillanatképállapottal rendelkezik, a folyamat meghibásodik .AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION A több frissítés pillanatképeinek feldolgozásához használd WITH VERSION (...).

    • VERZIÓVAL (version_query)

      Optional. Egy lekérdezés, amely kiválasztja a következő snapshot verziót a feldolgozáshoz. Pontosan egy sorra rendelhető oszlopot kell visszaadnia, és 0 vagy 1 sort. Amikor 1 sort ad vissza, az értéknek nem nullnak kell lennie. Az oszlop lehet skalárérték, például a BIGINT, vagy a STRUCT , amelynek mezői mind rendezhetők. Egy verziós lekérdezés, amely több oszlopot, több sort vagy null értéket ad vissza, megbukik az áramlásban .INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY

      Egy pipeline frissítés során a motor a következő lépéseket ismétli: értékeli a verzió lekérdezést; ha a lekérdezés 0 sort ad vissza, abbahagyja ennek a folyamatnak a feldolgozását a jelenlegi frissítésnél; ha a lekérdezés 1 sort ad vissza, a motor current_snapshot_version()-en keresztül mutatja be az értéket, értékeli a snapshot lekérdezést, commit-olja a kapott snapshotot, és a kötelezett verziót last_snapshot_version()-en keresztül mutatja be. Ezután a motor újraértékeli a verziós lekérdezést, hogy kiválassza a következő verziót. Egy egyetlen pipeline frissítés sorrendben dolgozza fel a verziókat, amíg a verziólekérdezés nem ad sorokat.

      Minden sikeres commit után visszahozott verziónak nagyobbnak kell lennie, mint a korábban elkötelezett verzió; egy nem növekvő verzió megbukja a frissítést .APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION A verzió értékének adattípusának változatlannak kell maradnia a snapshot commit-ek között; egy adattípus-változtatás meghibásodik a frissítésnél .AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED A teljes frissítés törli a tartós verzió állapotát.

    • KULCSOK

      Required. Az elsődleges kulcsoszlopokat a pillanatképek közötti sorok azonosítására használják a változásészlelés érdekében.

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

      Optional. Megadja, hogyan tárolják a változásokat a céltáblában. Az alapértelmezett érték a SCD TYPE 1.

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

      Optional. Csak akkor érvényes, ha SCD TYPE 2. Megadja, mely oszlopok indítanak el új történetsort, amikor azok változnak. Vagy adj egy explicit oszloplistát, vagy * EXCEPT (col_list) minden oszlopot kövess, kivéve a felsoroltokat.

    A Snapshot CDC nem támogatja WHERE az vagy SEQUENCE BY. A keresztpillanatkép sorrendezése a .WITH VERSION (...)

  • target_table

    A frissíteni kívánt táblázat. Ennek streamelésre alkalmas táblának kell lennie.

  • INSERT BE

    A céltáblába való beszúrásra szánt táblalekérdezést definiál. Ha a ONCE beállítás nincs megadva, a lekérdezésnek streamelési lekérdezésnek kell lennie. A STREAM kulcsszóval stream-szemantikát használhat a forrásból való olvasáshoz. Ha az olvasás egy meglévő rekord módosítását vagy törlését tapasztalja, hibaüzenet jelenik meg. A legbiztonságosabb, ha statikus vagy csak hozzáfűző forrásokból olvas. A módosítási véglegesítéseket tartalmazó adatok betöltéséhez használhatja a Pythont és a skipChangeCommits hibák kezelésére szolgáló lehetőséget.

    INSERT INTO kölcsönösen kizárja a AUTO CDC ... INTO. Akkor használható AUTO CDC ... INTO , ha a forrásadatok módosítási adatrögzítési (CDC) funkciót tartalmaznak. A INSERT INTO elemet akkor használja, ha a forrásban nem szerepel.

    További információért az adatfolyamokról, tekintse meg a Folyamatokkal történő adatátalakítást.

  • CSERÉLD KIHASZNÁLVA ( column_name [, ...] ) SOROZAT sequence_column

    Important

    Ez a funkció bétaverzióban érhető el. Databricks Runtime 18.2 vagy annál magasabb verziót igényel.

    Definiálja az áramlást, REPLACE USING amely a céltáblában lévő összes sort lecseréli, hogy illeszkedjen a megadott kulcsoszlopokhoz, és minden más sort érintetlenül hagyja. Használd, REPLACE USING ha a forrásod részleges pillanatképek sorozata, amelyeket oszlopok szerint kulcsolnak. SEQUENCE BY Rendeli a frissítéseket, így a kulcs legmagasabb sorrendje nyer, még akkor is, ha a frissítések sorrendben érkeznek.

    Legalább egy kulcsoszlopot és pontosan egy SEQUENCE BY oszlopot kell megadni. A lekérdezésnek streaming lekérdezésnek kell lennie, és BY NAME kötelező. REPLACE USING Nem lehet kombinálni vagy ONCE kombinálni AUTO CDC ... INTO.

    További információért lásd: Részleges snapshot helyettesítés CSERÉLJ USING flow-okkal.

  • CSERE WHERE állapot

    Definiálja a flow-t egy REPLACE WHERE áramlásként, amely újraszámolja és felülírja a céltáblák egy célzott részhalmazát. Minden frissítéskor a céltábla egyeztetésének condition minden sorát törlik, a forráslekérdezést ugyanarra a predikátumtartományra újraszámításra használják, és az eredményeket beillesztik. Azok a sorok, amelyek nem egyeznek condition , érintetlenül maradnak. Nem kell hozzáadni a predikátumot a forráslekérdezéshez; a pipeline engine automatikusan alkalmazza, amikor a forrásból olvas.

    REPLACE WHERE Batch szemantikát használ, így a forráslekérdezésnek nem kell streaming lekérdezésnek lennie. A BY NAME használata kötelező. REPLACE WHERENem kombinálható , ONCEREPLACE USING, vagy AUTO CDC ... INTO.

    További információért lásd: Köteg feldolgozás REPLACE WHERE flow-okkal és REPLACE WHERE flows önálló streaming táblákhoz.

  • EGYSZER

    Igény szerint a folyamatot egyszeri folyamatként, például visszatöltésként definiálhatja. A ONCE használata kétféleképpen változtatja meg a folyamatot:

    • A forrás query vagy create_auto_cdc_flow_spec nem egy folyamatos tábla.
    • A folyamat alapértelmezés szerint egyszer fut. Ha a folyamat teljes frissítéssel frissül, akkor a ONCE folyamat újra fut az adatok újbóli létrehozásához.

    ONCE Nem használható , REPLACE USINGami streaming forrást igényel.

Példák

-- 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;

További információkért az egyszeri visszatöltés és a folyamatban lévő CDC kombinálásáról ugyanazon célponton, lásd: Backfilling historical data pipelines-szal.

  • Utolsó frissítés időpontja: