SKAPA FLÖDE (pipelines)

Använd -instruktionen CREATE FLOW för att skapa flöden eller återfyllnad för tabeller i en pipeline.

Syntax

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

Parameterar

  • flow_name

    Namnet på flödet som ska skapas.

  • KOMMENTAR

    En valfri beskrivning av flödet.

  • AUTOMATISK CDC IN

    En AUTO CDC ... INTO -instruktion som definierar flödet med en create_auto_cdc_flow_spec. Du måste antingen inkludera en AUTO CDC ... INTO -instruktion eller en INSERT INTO -instruktion. Använd AUTO CDC ... INTO när källfrågan använder ändra datasemantik.

    Mer information finns i AUTO CDC INTO (pipelines).

  • target_table

    Tabellen som ska uppdateras. Detta måste vara en strömningstabell.

  • INSERT IN

    Definierar en tabellfråga som infogas i måltabellen. Om alternativet ONCE inte anges måste frågan vara en direktuppspelningsfråga . Använd stream-nyckelordet för att använda strömmande semantik för att läsa från källan. Om läsningen påträffar en ändring eller borttagning av en befintlig post utlöses ett fel. Det är säkrast att läsa från statiska eller endast tilläggskällor. Om du vill mata in data som har ändringsincheckningar kan du använda Python och skipChangeCommits alternativet för att hantera fel.

    INSERT INTO kan inte kombineras med AUTO CDC ... INTO. Använd AUTO CDC ... INTO när källdata innehåller CDC-funktionalitet (change data capture). Använd INSERT INTO när källan inte gör det.

    Mer information om strömmande data finns i Transformera data med pipelines.

  • BYT UT MED ( column_name [, ...] ) SEKVENS AV sequence_column

    Important

    Den här funktionen finns i Beta. Kräver Databricks Runtime 18.2 och senare.

    Definierar flödet som ett REPLACE USING flöde, vilket ersätter alla rader i måltabellen som matchar de angivna nyckelkolumnerna och lämnar alla andra rader orörda. Använd REPLACE USING när din källa är en serie partiella snapshots som är kodade efter kolumn. SEQUENCE BY ordnar uppdateringarna så att den högsta sekvensen för en nyckel vinner, även när uppdateringar kommer i fel ordning.

    Ange minst en nyckelkolumn och exakt en SEQUENCE BY kolumn. Frågan måste vara en strömmande fråga och BY NAME är obligatorisk. REPLACE USING kan inte kombineras med ONCE eller med AUTO CDC ... INTO.

    För mer information, se Partial snapshot replacement med REPLACE USING flows.

  • En gång

    Du kan också definiera flödet som ett engångsflöde, till exempel en återfyllnad. Om du använder ONCE ändras flödet på två sätt:

    • Källan query eller create_auto_cdc_flow_spec är inte en strömmande tabell.
    • Flödet körs en gång som standard. Om pipelinen uppdateras med en komplett uppdatering, så körs flödet igen för att återskapa data.

    ONCE kan inte användas med REPLACE USING, vilket kräver en strömningskälla.

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