CREATE TABLE ... FLOW (Pipelines)

Important

Dieses Feature befindet sich in der Betaversion.

Verwenden Sie die CREATE TABLE ... FLOW Anweisung, um eine verwaltete Tabelle in einer Pipeline zu erstellen, die von einem oder mehreren Flüssen geschrieben wurde.

Syntax

CREATE TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ flow_clause ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
        [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...] )

table_clauses
  { PARTITIONED BY (col [, ...]) |
    CLUSTER BY clause |
    LOCATION path |
    COMMENT table_comment |
    TBLPROPERTIES clause |
    WITH { ROW FILTER clause } } [ ... ]

flow_clause
  FLOW [ IMPORT CHECKPOINT checkpoint_path ] INSERT [ONCE] BY NAME query

Um mehrere Quellen in eine verwaltete Tabelle einzufächern, deklarieren Sie mehrere Flüsse, die auf sie mit CREATE FLOW (Pipelines) abzielen:

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parameters

  • table_name

    Der Name der zu erstellenden verwalteten Tabelle. Wenn der Name nicht qualifiziert ist, wird die Tabelle im Zielschema der Pipeline erstellt. Der Name darf noch nicht zu einer Streamingtabelle gehören.

  • Tabellenspezifikation

    Definiert optional die Spalten, deren Typen, Eigenschaften und Beschreibungen. Wenn sie weggelassen wird, wird das Schema aus der Flussabfrage abgeleitet.

  • CONSTRAINT expectation_name ERWARTEN (expectation_expr) [ ON VIOLATION { FAIL UPDATE | DROP ROW } ]

    Fügt Datenqualitätserwartungen in die verwaltete Tabelle hinzu. Diese Erwartungen an die Datenqualität können zeitlich verfolgt und über das Ereignisprotokoll der Pipeline abgerufen werden. Die Erwartung FAIL UPDATE bewirkt, dass die Verarbeitung nicht erfolgreich ist, wenn die Tabelle sowohl erstellt als auch aktualisiert wird. Die Erwartung DROP ROW bewirkt, dass die gesamte Zeile gelöscht wird, wenn die Erwartung nicht erfüllt wird. Weitere Informationen finden Sie unter Verwalten der Datenqualität mit Pipelineerwartungen.

    expectation_expr können aus Literalen, Spaltenkennungen innerhalb der Tabelle und deterministischen, integrierten SQL-Funktionen oder -Operatoren bestehen, außer:

    Darüber hinaus darf expectation_expr keine Unterabfrage enthalten.

  • PARTITIONIERT VON (Spalte [, ...])

    Partitioniert die Tabelle optional durch eine Teilmenge von Spalten.

  • CLUSTER BY Klausel

    Ermöglicht optional die Flüssigkeitsclusterung auf dem Tisch. Sie können nicht kombinieren PARTITIONED BY und CLUSTER BY.

  • LOCATION-Pfad

    Ein optionaler Speicherort für die Tabellendaten.

  • KOMMENTAR table_comment

    Ein STRING Literal, das die Tabelle beschreibt.

  • TBLPROPERTIES-Klausel

    Legt optional eine oder mehrere benutzerdefinierte Tabelleneigenschaften fest.

  • WITH-Klausel ROW FILTER

    Fügt der Tabelle eine Zeilenfilterfunktion hinzu. Zukünftige Abfragen für diese Tabelle erhalten eine Teilmenge der Zeilen, für die die Funktion ausgewertet wird TRUE.

  • FLOW INSERT [ONCE] BY NAME-Abfrage

    Definiert einen Anfügefluss, der das Ergebnis der query Tabelle einfügt, wobei Ergebnisspalten nach Namen zu Tabellenspalten übereinstimmen. query kann auf Batch- oder Streamingquellen verweisen. ONCE führt den Ablauf nur einmal aus (z. B. bei einem Rückfüllvorgang), anstatt auf jeder Aktualisierung. Jeder benannte Fluss verarbeitet seine Eingabe genau einmal pro Pipelineaktualisierung, identisch mit FLOW INSERT BY NAME einer Streamingtabelle.

  • IMPORT-KONTROLLPUNKT checkpoint_path

    Ein STRING Literal mit dem Pfad zu einem bestehenden Structured Streaming-Checkpoint, der in den Flow importiert wird, sodass ein migrierter Stream von seinem letzten Committed Offset fortgesetzt wird, anstatt die Quelle neu zu verarbeiten. Das Importieren eines Kontrollpunkts befindet sich in der Beta. Siehe Migrate a Structured Streaming Checkpoint.

Migrieren eines strukturierten Streaming-Checkpoints

Important

Das Importieren eines Kontrollpunkts befindet sich in der Beta.

Nutze IMPORT CHECKPOINT es, um eine bestehende Structured Streaming-Workload in eine Pipeline zu migrieren, ohne den Quellcode neu zu verarbeiten. Stellen Sie checkpoint_path die checkpointLocation verwendete Structured Streaming-Abfrage ein, die ein Cloud-Speicher, ein Unity-Catalog-Volume oder ein DBFS-Pfad sein kann. Beim ersten Pipeline-Update klont der Flow diesen Checkpoint in den verwalteten Speicher der Pipeline. Der Fluss wird dann mit seinem Zustand (wie Aggregationen, Deduplizierungsschlüsseln und Wasserzeichen) vom zuletzt zugesagten Offset fortgesetzt. Nachfolgende Pipeline-Aktualisierungen verwenden den geklonten Checkpoint des Flows; der ursprüngliche Checkpoint wird nicht verändert.

Warning

Stoppen Sie die ursprüngliche Structured Streaming-Abfrage, bevor Sie die Pipeline ausführen. Die ursprüngliche Structured Streaming-Abfrage kann nach dem Import wiederverwendet werden, aber Sie müssen ihren Checkpoint-Status verwalten und sicherstellen, dass Pipeline und Abfrage nicht gleichzeitig in dieselbe Tabelle schreiben, was doppelte Daten erzeugen kann.

Erstelle die Structured Streaming-Abfrage als Pipeline-Fluss, der weiterhin auf die Tabelle schreibt, in die die Anfrage geschrieben hat, und importiert ihren Checkpoint:

-- Use the existing table your Structured Streaming query writes to; CREATE TABLE adopts it.
CREATE TABLE streaming_table
-- Continue from the imported checkpoint instead of reprocessing the source.
FLOW IMPORT CHECKPOINT '/Volumes/my_catalog/my_schema/checkpoints/my_stream'
INSERT BY NAME
  -- The same source your original query read from.
  SELECT * FROM STREAM(source_table);

Der Checkpoint wird nur beim ersten Pipeline-Update importiert. Spätere Updates ignorieren IMPORT CHECKPOINT. Eine vollständige Aktualisierung importiert den Checkpoint nicht erneut. Es beginnt von einem neuen, leeren Kontrollpunkt und verarbeitet die Quelle erneut.

Einschränkungen

  • Verwaltete Tabellen unterstützen keine CDC-Änderungsflüsse. AUTO CDC INTO(SQL) oder apply_changes / apply_changes_from_snapshot (Python) für eine verwaltete Tabelle schlägt mit MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. Für CDC-Ziele verwenden Sie a CREATE STREAMING TABLE (Pipelines) mit einem Flow, der durch CREATE FLOW (Pipelines) definiert ist.
  • Verwaltete Tabellen werden nicht unterstützt FLOW ... REPLACE WHERE. Nur FLOW INSERT BY NAME wird unterstützt. Um eine gezielte Teilmenge einer Tabelle neu zu berechnen und zu überschreiben, erstellen Sie eine Streaming-Tabelle und definieren Sie einen REPLACE WHERE Fluss mit CREATE FLOW (Pipelines). Siehe Batchverarbeitung mit REPLACE-FlüssenWHERE.
  • Verwaltete Tabellen werden nur in Pipelines mit Unity-Katalog unterstützt. Der Hive-Metaspeicher wird nicht unterstützt.
  • Sie können den Namen einer vorhandenen Streamingtabelle für eine verwaltete Tabelle nicht wiederverwenden. Legen Sie die Streamingtabelle zuerst ab, oder die Anweisung schlägt mit CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLE.
  • IMPORT CHECKPOINT wird nur auf FLOW INSERT BY NAMEunterstützt.

Beispiele

-- Create a managed table populated by an inline append flow from a streaming table
CREATE TABLE output
FLOW INSERT BY NAME SELECT * FROM STREAM(samples.tpch.orders);

-- Create a managed table that ingests files with schema inference and evolution
CREATE TABLE raw_data
FLOW INSERT BY NAME
  SELECT * FROM STREAM read_files('abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/base/path');

-- Create a partitioned managed table from a streaming source
CREATE TABLE events
PARTITIONED BY (bucket)
FLOW INSERT BY NAME
  SELECT id, bucket FROM STREAM read_files('abfss://my_path', format => 'json');

-- Create a managed table with liquid clustering
CREATE TABLE orders_clustered
CLUSTER BY (order_date, customer_id)
FLOW INSERT BY NAME
  SELECT
    o_orderkey   AS order_id,
    o_custkey    AS customer_id,
    o_orderdate  AS order_date,
    o_totalprice AS total_price
  FROM STREAM(samples.tpch.orders);

-- Create a managed table with a data quality expectation that drops violating rows
CREATE TABLE valid_events
  (CONSTRAINT positive_id EXPECT (id > 0) ON VIOLATION DROP ROW)
FLOW INSERT BY NAME
  SELECT id FROM STREAM read_files('s3://bucket/path', format => 'json');

Weitere Ressourcen

  • Zuletzt aktualisiert um