CREATE FLOW (Pipelines)

Verwenden Sie die CREATE FLOW Anweisung, um Datenflüsse oder Rückfüllungen für Tabellen in einer Pipeline zu erstellen.

Note

CREATE FLOW Wenn man auf eine Streaming-Tabelle zielt, unterstützt es sowohl AUTO CDC ... INTO als auch REPLACE WHERE Flows. Eine verwaltete Tabelle, die mit CREATE TABLE ... FLOW erstellt wird, unterstützt keine Erfassung von Änderungsdaten: Ein AUTO CDC ... INTO Flow gegen eine verwaltete Tabelle schlägt mit MANAGED_TABLE_DOES_NOT_SUPPORT_CDCfehl. CREATE FLOW In eine Streaming-Tabelle unterliegt dieser Einschränkung nicht. Siehe CREATE TABLE ... FLOW (Pipelines).

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

Die Parameter

  • flow_name

    Der Name des zu erstellenden Flusses.

  • KOMMENTAR

    Eine optionale Beschreibung für den Fluss.

  • AUTOMATISCHE CDC IN

    Eine AUTO CDC ... INTO Anweisung, die den Ablauf mithilfe einer create_auto_cdc_flow_spec definiert. Sie müssen entweder eine AUTO CDC ... INTO Anweisung oder eine INSERT INTO Anweisung einschließen. Verwenden Sie AUTO CDC ... INTO bei Abfragen, wenn die Quellabfrage die Änderungsdatensemantik nutzt.

    Weitere Informationen finden Sie unter AUTO CDC INTO (Pipelines).

  • AUTO CDC ... AUS SNAPSHOT

    Eine Aussage, die Änderungen durch Vergleich von Snapshots AUTO CDC ... INTO ableitet, anstatt einen Änderungsfeed zu lesen. Verwenden Sie dieses Formular, wenn die Datenerfassung von Änderungen im Quellcode nicht aktiviert ist und nur vollständige Schnappschüsse verfügbar sind. Die Quelle ist in zwei Teile spezifiziert: eine erforderliche FROM SNAPSHOT (snapshot_query) Klausel, die die Snapshot-Daten liest, und eine optionale WITH VERSION (version_query) Klausel, die die nächste Snapshot-Version auswählt. Erfahren Sie , wie AUTO CDC FROM SNAPSHOT funktioniert.

    • AUS SNAPSHOT (snapshot_query)

      Required. Eine Abfrage, die die Snapshot-Daten für die von der von ausgewählte Version ausliest WITH VERSION (...). Die Engine diffifiziert das Ergebnis mit dem zuvor festgelegten Snapshot, um Einfügungen, Aktualisierungen und Löschungen abzuleiten, und führt diese mit der Zeilenidentität in das Ziel KEYS zusammen und STORED AS bestimmt, wie Änderungen gespeichert werden.

      Rufen current_snapshot_version() in dieser Abfrage auf, um auf die von ausgewählte WITH VERSION (...)Version zuzugreifen. Wenn WITH VERSION (...) nicht spezifiziert ist, current_snapshot_version() ist innerhalb nicht aufrufbar.FROM SNAPSHOT (...)

      Wenn WITH VERSION (...) weggelassen wird, liest die Engine die Quellcode direkt durch FROM SNAPSHOT (...), und die Snapshot-Abfrage wird nur während der Erstladezeit ausgeführt, während das Ziel keine zugeteilten Daten und keinen zugesagten Snapshot-Status hat. Bei jeder späteren Aktualisierung, wenn das Ziel bereits Daten enthält oder einen Snapshot-Status hat, scheitert der Fluss mit AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION. Um Snapshots über mehrere Updates hinweg zu verarbeiten, verwenden WITH VERSION (...)Sie .

    • MIT VERSION (version_query)

      Optional. Eine Abfrage, die die nächste Snapshot-Version auswählt, die verarbeitet wird. Es muss genau eine Spalte eines ordnenden Typs und entweder 0 oder 1 Zeile zurückgeben. Wenn sie 1 Zeile zurückgibt, muss der Wert nicht null sein. Die Spalte kann ein skalarer Wert sein, wie zum Beispiel a BIGINT, oder a STRUCT , dessen Felder alle ordnbar sind. Eine Versionsabfrage, die mehr als eine Spalte, mehr als eine Zeile oder einen Nullwert zurückgibt, scheitert im Fluss mit INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY.

      Während eines Pipeline-Updates wiederholt die Engine folgende Schritte: Sie bewertet die Versionsabfrage; wenn die Abfrage 0 Zeilen zurückgibt, stoppt sie die Verarbeitung dieses Flows für das aktuelle Update; wenn die Abfrage 1 Zeile zurückgibt, stellt die Engine diesen Wert über current_snapshot_version(), bewertet die Snapshot-Abfrage, commit das resultierende Snapshot und stellt die committierte Version über last_snapshot_version(). Die Engine bewertet dann die Versionsabfrage erneut, um die nächste Version auszuwählen. Ein einzelnes Pipeline-Update verarbeitet Versionen der Reihe nach, bis die Versionsabfrage keine Zeilen mehr zurückgibt.

      Jede nach einem erfolgreiche Commit zurückgegebene Version muss größer sein als die zuvor committierte Version; eine nicht wachsende Version scheitert bei der Aktualisierung mit APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION. Der Datentyp des Versionswerts muss über Snapshot-Comits hinweg unverändert bleiben; eine Änderung des Datentyps scheitert bei der Aktualisierung mit AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Eine vollständige Aktualisierung löscht den Status der persistierten Version.

    • SCHLÜSSEL

      Required. Die Primärschlüsselspalten werden verwendet, um Zeilen über Snapshots hinweg zur Änderungserkennung zu identifizieren.

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

      Optional. Spezifiziert, wie Änderungen in der Zieltabelle gespeichert werden. Der Standardwert lautet SCD TYPE 1.

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

      Optional. Gilt nur mit SCD TYPE 2. Spezifiziert, welche Spalten eine neue Historik-Zeile auslösen, wenn sie sich ändern. Stellen Sie entweder eine explizite Spaltenliste bereit oder * EXCEPT (col_list) verfolgen Sie jede Spalte außer den aufgeführten.

    Snapshot CDC unterstützt WHERE keine oder SEQUENCE BY. Die Kreuz-Snapshot-Ordnung wird durch WITH VERSION (...)ausgedrückt.

  • target_table

    Die Tabelle, die aktualisiert werden soll. Dies muss eine Streamingtabelle sein.

  • INSERT IN

    Definiert eine Tabellenabfrage, die in die Zieltabelle eingefügt wird. Wenn die Option nicht angegeben wird, muss es sich bei der ONCE Abfrage um eine Streamingabfrage handeln. Verwenden Sie das STREAM-Schlüsselwort, um Streamingsemantik zum Lesen aus der Quelle zu verwenden. Wenn beim Lesen eine Änderung oder löschung in einem vorhandenen Datensatz auftritt, wird ein Fehler ausgelöst. Es ist am sichersten, aus statischen oder nur angefügten Quellen zu lesen. Zum Einlesen von Daten mit Änderungs-Commits können Sie Python und die skipChangeCommits Option zur Fehlerbehandlung verwenden.

    INSERT INTO schließt sich mit AUTO CDC ... INTO gegenseitig aus. Verwenden Sie diese Funktion AUTO CDC ... INTO , wenn die Quelldaten die Funktionalität für die Änderungsdatenerfassung (Change Data Capture, CDC) enthalten. Verwenden Sie INSERT INTO, wenn die Quelle nicht verwendet wird.

    Weitere Informationen zum Streamen von Daten finden Sie unter Transformieren von Daten mit Pipelines.

  • ERSETZEN MIT ( column_name [, ...] ) SEQUENZ VON sequence_column

    Important

    Dieses Feature befindet sich in der Betaversion. Benötigt Databricks Runtime 18.2 und höher.

    Definiert den Fluss als Fluss REPLACE USING , der alle Zeilen in der Zieltabelle ersetzt, die mit den angegebenen Schlüsselspalten übereinstimmen, und alle anderen Zeilen unberührt lässt. Verwenden Sie REPLACE USING , wenn Ihre Quelle eine Reihe von teilweisen Schnappschüssen ist, die nach Spalten verschlüsselt sind. SEQUENCE BY ordnet die Updates so, dass die höchste Reihenfolge eines Schlüssels gewinnt, selbst wenn Updates nicht in der richtigen Reihenfolge eintreffen.

    Geben Sie mindestens eine Schlüsselspalte und genau eine SEQUENCE BY Spalte an. Die Abfrage muss eine Streaming-Abfrage sein und BY NAME ist erforderlich. REPLACE USING kann weder mit ONCE noch mit AUTO CDC ... INTOkombiniert werden.

    Weitere Informationen finden Sie unter Teilweise Snapshot-Ersetzung mit ERSETZEN UNTER Verwendung von Flows.

  • ERSETZE-Bedingung WHERE

    Definiert den Fluss als einen REPLACE WHERE Fluss, der eine Ziel-Teilmenge der Zieltabelle neu berechnet und überschreibt. Bei jedem Update werden alle Zeilen im Zieltabellenabgleich condition gelöscht, die Quellanfrage für denselben Prädikatbereich neu berechnet und die Ergebnisse eingefügt. Reihen, die nicht übereinstimmen condition , bleiben unberührt. Du musst das Prädikat nicht zur Quellanfrage hinzufügen; die Pipeline-Engine verwendet es automatisch beim Lesen aus der Quelle.

    REPLACE WHERE verwendet Batch-Semantik, sodass die Quellanfrage keine Streaming-Abfrage sein muss. BY NAME ist erforderlich. REPLACE WHERE Kann nicht mit ONCE, REPLACE USINGoder AUTO CDC ... INTOkombiniert werden.

    Weitere Informationen finden Sie unter Batch-Verarbeitung mit ERSETZEN-Flows WHERE und REPLACE WHERE Flows für eigenständige Streaming-Tabellen.

  • EINMAL

    Definieren Sie optional den Fluss als einmaligen Ablauf, z. B. als Rückfüllvorgang. Die Verwendung von ONCE ändert den Ablauf auf zwei Arten:

    • Die Quelle query oder create_auto_cdc_flow_spec ist keine Streamingtabelle.
    • Der Fluss wird standardmäßig einmal ausgeführt. Wenn die Pipeline mit einer vollständigen Aktualisierung aktualisiert wird, wird der ONCE Flow erneut ausgeführt, um die Daten neu zu erstellen.

    ONCE Kann nicht mit REPLACE USINGverwendet werden, was eine Streaming-Quelle erfordert.

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;

Weitere Informationen zur Kombination eines einmaligen Backfill mit laufendem CDC auf demselben Ziel finden Sie unter Backfilling historische Daten mit Pipelines.

  • Zuletzt aktualisiert um