CREATE FLOW (pijplijnen)

Gebruik de CREATE FLOW instructie om gegevensstromen of backfills te maken voor tabellen in een pijplijn.

Note

CREATE FLOW Gericht op een streamingtabel ondersteunt zowel AUTO CDC ... INTO als REPLACE WHERE flows. Een beheerde tabel die met CREATE TABLE ... FLOW is aangemaakt, ondersteunt geen wijzigingsdata-vastlegging: een AUTO CDC ... INTO flow tegen een beheerde tabel faalt met MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. CREATE FLOW In een streamingtabel valt niet onder die beperking. Zie CREATE TABLE ... FLOW (pijpleidingen).

Syntaxis

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

Parameterwaarden

  • flow_name

    De naam van de werkstroom die moet worden aangemaakt.

  • COMMENTAAR

    Een optionele beschrijving voor de stroom.

  • AUTOMATISCHE CDC IN

    Een AUTO CDC ... INTO instructie die de flow definieert, met een create_auto_cdc_flow_spec. Je moet een AUTO CDC ... INTO verklaring of een INSERT INTO verklaring opnemen. Gebruik AUTO CDC ... INTO wanneer de bronquery gegevenssemantiek voor wijzigingen hanteert.

    Zie AUTO CDC INTO (pijplijnen) voor meer informatie.

  • AUTO CDC ... VAN SNAPSHOT

    Een AUTO CDC ... INTO statement dat veranderingen afleidt door snapshots te vergelijken in plaats van een wijzigingsfeed te lezen. Gebruik dit formulier wanneer wijzigingsgegevensopname niet is ingeschakeld op de bron en alleen volledige snapshots beschikbaar zijn. De bron is gespecificeerd in twee delen: een vereiste FROM SNAPSHOT (snapshot_query) clausule die de snapshotgegevens leest, en een optionele WITH VERSION (version_query) clausule die de volgende snapshotversie selecteert om te verwerken. Zie hoe AUTO CDC FROM SNAPSHOT werkt.

    • VAN SNAPSHOT (snapshot_query)

      Required. Een query die de snapshotgegevens leest voor de versie die door WITH VERSION (...)is geselecteerd. De engine differenteert het resultaat met de eerder toegewijde snapshot om inserts, updates en deletes af te leiden, en voegt deze samen met het doel met behulp KEYS van for row identity en STORED AS om te bepalen hoe wijzigingen worden opgeslagen.

      Roep current_snapshot_version() aan binnen deze query om te refereren naar de versie die door geselecteerd is WITH VERSION (...). Als WITH VERSION (...) niet gespecificeerd is, current_snapshot_version() is niet aanroepbaar binnen FROM SNAPSHOT (...).

      Wanneer WITH VERSION (...) wordt weggelaten, leest de engine de bron direct door FROM SNAPSHOT (...), en de snapshot-query wordt alleen uitgevoerd tijdens de initiële lading, terwijl het doelwit geen toegewezen data en geen toegewijde snapshotstatus heeft. Bij elke latere update, wanneer het doel al data bevat of een snapshot-status heeft toegewijd, faalt de flow met AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION. Om snapshots te verwerken over meerdere updates, gebruik WITH VERSION (...).

    • MET VERSIE (version_query)

      Optional. Een query die de volgende snapshotversie selecteert om te verwerken. Het moet precies één kolom van een ordegoed type teruggeven en ofwel 0 of 1 rij. Wanneer het 1 rij teruggeeft, moet de waarde niet-nul zijn. De kolom kan een scalaire waarde zijn, zoals een BIGINT, of een STRUCT waarvan de velden allemaal ordenbaar zijn. Een versiequery die meer dan één kolom, meer dan één rij of een nullwaarde teruggeeft, faalt de flow met INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY.

      Tijdens een pipeline-update herhaalt de engine de volgende stappen: hij evalueert de versiequery; als de query 0 rijen teruggeeft, stopt hij met het verwerken van deze flow voor de huidige update; als de query 1 rij teruggeeft, stelt de engine die waarde bloot via current_snapshot_version(), evalueert de snapshot-query, commit de resulterende snapshot en stelt de gecommitteerde versie bloot via last_snapshot_version(). De engine evalueert vervolgens opnieuw de versiequery om de volgende versie te selecteren. Een enkele pipeline-update verwerkt versies in volgorde totdat de versiequery geen rijen meer teruggeeft.

      Elke versie die na een succesvolle commit wordt teruggegeven, moet groter zijn dan de eerder toegewezen versie; een niet-toenemende versie faalt de update met APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION. Het datatype van de versiewaarde moet ongewijzigd blijven over snapshot-commits; een wijziging van het datatype faalt bij de update met AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Een volledige verversing verwijdert de status van de persisted versie.

    • SLEUTELS

      Required. De primaire sleutelkolommen worden gebruikt om rijen over snapshots heen te identificeren voor wijzigingsdetectie.

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

      Optional. Specificeert hoe wijzigingen worden opgeslagen in de doeltabel. De standaardwaarde is SCD TYPE 1.

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

      Optional. Alleen van toepassing met SCD TYPE 2. Geeft aan welke kolommen een nieuwe geschiedenisrij activeren wanneer ze veranderen. Geef ofwel een expliciete kolomlijst of * EXCEPT (col_list) volg elke kolom behalve de vermelde.

    Snapshot CDC ondersteunt WHERE geen of SEQUENCE BY. Cross-snapshot ordening wordt uitgedrukt via WITH VERSION (...).

  • target_table

    De tabel die moet worden bijgewerkt. Dit moet een Streaming tabel zijn.

  • INSERT IN

    Hiermee definieert u een query voor een tabel die wordt ingevoegd in de doeltabel. Als de ONCE optie niet is opgegeven, moet de query een streamingquery zijn. Gebruik het trefwoord STREAM om streaming-semantiek te gebruiken om uit de bron te lezen. Als de leesbewerking een wijziging of verwijdering van een bestaande record tegenkomt, wordt er een fout gegenereerd. Het is het veiligst om te lezen uit statische of alleen bij te voegen bronnen. Als u gegevens wilt opnemen die wijzigingen doorvoeren, kunt u Python en de skipChangeCommits optie gebruiken om fouten te verwerken.

    INSERT INTO is wederzijds exclusief met AUTO CDC ... INTO. Gebruik AUTO CDC ... INTO wanneer de brongegevens de CDC-functionaliteit (Change Data Capture) bevatten. Gebruik INSERT INTO deze functie wanneer de bron dat niet doet.

    Zie Gegevens transformeren met pijplijnenvoor meer informatie over het streamen van gegevens.

  • VERVANG MET ( column_name [, ...] ) VOLGORDE DOOR sequence_column

    Important

    Deze functie bevindt zich in de bètaversie. Vereist Databricks Runtime 18.2 en hoger.

    Definieert de flow als een REPLACE USING flow, die alle rijen in de doeltabel vervangt die overeenkomen met de gespecificeerde sleutelkolommen en alle andere rijen onaangeroerd laat. Gebruik REPLACE USING wanneer je bron een reeks gedeeltelijke snapshots is die per kolom zijn gecodeerd. SEQUENCE BY ordent de updates zodat de hoogste volgorde voor een sleutel wint, zelfs wanneer updates buiten volgorde binnenkomen.

    Specificeer ten minste één sleutelkolom en precies één SEQUENCE BY kolom. De query moet een streaming-query zijn en BY NAME is verplicht. REPLACE USING Kan niet gecombineerd worden met ONCE of met AUTO CDC ... INTO.

    Voor meer informatie, zie Gedeeltelijke snapshotvervanging met VERVANG MET stromen.

  • VERVANG-voorwaarde WHERE

    Definieert de flow als een REPLACE WHERE flow, die een doelgerichte deelverzameling van de doeltabel herberekent en overschrijft. Bij elke update worden alle rijen in de doeltabelmatching condition verwijderd, wordt de bronquery opnieuw berekend voor datzelfde predicaatbereik en worden de resultaten ingevoegd. Rijen die niet overeenkomen condition blijven onaangeroerd. Je hoeft het predicaat niet toe te voegen aan de bronquery; de pipeline-engine past het automatisch toe bij het lezen van de bron.

    REPLACE WHERE gebruikt batchsemantiek, dus de bronquery hoeft geen streamingquery te zijn. BY NAME is vereist. REPLACE WHERE kan niet gecombineerd worden met ONCE, REPLACE USING, of AUTO CDC ... INTO.

    Voor meer informatie, zie Batchverwerking met VERVANG WHERE stromen en VERVANG WHERE stromen voor zelfstandige stroomtabellen.

  • EENMAAL

    U kunt de stroom desgewenst definiëren als een eenmalige stroom, zoals een backfill. Door ONCE te gebruiken verandert de stroom op twee manieren:

    • De bron query of create_auto_cdc_flow_spec is geen streamingtabel.
    • De stroom wordt standaard één keer uitgevoerd. Als de pijplijn wordt bijgewerkt met een volledige vernieuwing, wordt de ONCE stroom opnieuw uitgevoerd om de gegevens opnieuw te maken.

    ONCE Kan niet gebruikt worden met REPLACE USING, waarvoor een streamingbron nodig is.

Voorbeelden

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

Voor meer informatie over het combineren van een eenmalige backfill met doorlopende CDC op hetzelfde doel, zie Backfilling historische data met pijpleidingen.

  • Laatst bijgewerkt om