SKAPA FLÖDE (pipelines)

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

Note

CREATE FLOW Att rikta in sig på en strömningstabell stöder både AUTO CDC ... INTO och REPLACE WHERE flöden. En hanterad tabell skapad med CREATE TABLE ... FLOW stöder inte ändringsdatainsamling: ett AUTO CDC ... INTO flöde mot en hanterad tabell misslyckas med MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. CREATE FLOW Att gå in i en streamingtabell omfattas inte av den begränsningen. Se CREATE TABLE ... FLÖDE (rörledningar).

Syntax

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

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

  • AUTO CDC ... FRÅN SNAPSHOT

    Ett AUTO CDC ... INTO påstående som härleder ändringar genom att jämföra ögonblicksbilder istället för att läsa ett ändringsflöde. Använd detta formulär när ändringsdatafångst inte är aktiverat på källkoden och endast fullständiga ögonblicksbilder är tillgängliga. Källkoden specificeras i två delar: en obligatorisk FROM SNAPSHOT (snapshot_query) klausul som läser snapshot-datan, och en valfri WITH VERSION (version_query) klausul som väljer nästa snapshot-version att bearbeta. Se Så här fungerar AUTO CDC FRÅN SNAPSHOT.

    • FRÅN SNAPSHOT (snapshot_query)

      Required. En fråga som läser snapshot-data för den version som valts av WITH VERSION (...). Motorn diffifierar resultatet mot den tidigare committerade snapshoten för att härleda insättningar, uppdateringar och borttagningar, och slår ihop dem med målet med hjälp av KEYS för radidentitet och STORED AS för att avgöra hur ändringar lagras.

      Anropa current_snapshot_version() i denna fråga för att referera till versionen som valts av WITH VERSION (...). Om inte WITH VERSION (...) specificeras current_snapshot_version() är inte kallbar inuti FROM SNAPSHOT (...).

      När WITH VERSION (...) utelämnas läser motorn källkoden direkt genom FROM SNAPSHOT (...), och snapshot-frågan körs endast under den initiala laddningen, medan målet inte har någon committerad data och inget committed snapshot-tillstånd. Vid varje senare uppdatering, när målet redan innehåller data eller har bundit snapshot-tillstånd, misslyckas flödet med AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION. För att bearbeta snapshots över flera uppdateringar, använd WITH VERSION (...).

    • MED VERSION (version_query)

      Optional. En fråga som väljer nästa snapshot-version att bearbeta. Den måste returnera exakt en kolumn av en ordnbar typ och antingen 0 eller 1 rader. När den returnerar 1 rad måste värdet vara icke-null. Kolumnen kan vara ett skalärt värde, såsom a BIGINT, eller a STRUCT vars fält alla är ordnade. En versionsfråga som returnerar mer än en kolumn, fler än en rad eller ett nullvärde misslyckas med flödet med INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY.

      Under en pipelineuppdatering upprepar motorn följande steg: den utvärderar versionsfrågan; om frågan returnerar 0 rader slutar den bearbeta detta flöde för den aktuella uppdateringen; om frågan returnerar 1 rad exponerar motorn det värdet genom current_snapshot_version(), utvärderar snapshot-frågan, commar den resulterande snapshoten och exponerar den committerade versionen via last_snapshot_version(). Motorn utvärderar sedan versionsfrågan igen för att välja nästa version. En enda pipelineuppdatering bearbetar versioner i ordning tills versionsfrågan inte ger några rader.

      Varje version som returneras efter en lyckad commit måste vara större än den tidigare committerade versionen; en icke-ökande version misslyckas med uppdateringen med APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION. Versionsvärdets datatyp måste förbli oförändrad över snapshot-commits; en ändring av datatyp misslyckas med uppdateringen med AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. En fullständig uppdatering rensar tillståndet för den bestående versionen.

    • NYCKLAR

      Required. Primärnyckelkolumnerna används för att identifiera rader över snapshots för ändringsdetektion.

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

      Optional. Specificerar hur ändringar lagras i måltabellen. Standardvärdet är SCD TYPE 1.

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

      Optional. Gäller endast med SCD TYPE 2. Specificerar vilka kolumner som triggar en ny historikrad när de ändras. Ange antingen en explicit kolumnlista eller * EXCEPT (col_list) för att spåra varje kolumn utom de listade.

    Snapshot CDC stöder WHERE inte eller SEQUENCE BY. Korsögonblicksordning uttrycks genom WITH VERSION (...).

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

  • ERSÄTT-villkoret WHERE

    Definierar flödet som ett REPLACE WHERE flöde, som omberäknar och skriver över en måldelmängd av måltabellen. Vid varje uppdatering raderas alla rader i måltabellens matchning condition , källfrågan beräknas om för samma predikatintervall och resultaten infogas. Rader som inte matchar condition lämnas orörda. Du behöver inte lägga till predikatet i källfrågan; pipeline-motorn tillämpar det automatiskt när den läser från källkoden.

    REPLACE WHERE använder batch-semantik, så källfrågan behöver inte vara en strömningsfråga. BY NAME måste anges. REPLACE WHERE kan inte kombineras med ONCE, REPLACE USING, eller AUTO CDC ... INTO.

    För mer information, se Batchbearbetning med VERVANG WHERE flöden och ERSÄTT WHERE flöden för fristående strömningstabeller.

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

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

För mer om att kombinera en engångsfyllning med pågående CDC på samma mål, se Återfyllning av historisk data med pipelines.

  • Uppdaterades senast