Freigeben über

CREATE STREAMING TABLE

  • Artikel

Gilt für:Grünes Häkchen für „Ja“ Databricks SQL

Erstellt eine Streamingtabelle. Das ist eine Delta-Tabelle mit zusätzlicher Unterstützung von Streaming oder inkrementeller Datenverarbeitung.

Streamingtabellen werden nur in Delta Live Tables und in Databricks SQL mit Unity Catalog unterstützt. Wenn dieser Befehl für unterstützte Databricks Runtime-Computeressourcen ausgeführt wird, wird nur die Syntax analysiert. Siehe Entwickeln von Pipelinecode mit SQL.

Syntax

{ CREATE OR REFRESH STREAMING TABLE | CREATE STREAMING TABLE [ IF NOT EXISTS ] }
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ AS query ]

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

column_properties
  { NOT NULL |
    COMMENT column_comment |
    column_constraint |
    MASK clause } [ ... ]

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

schedule_clause
  { EVERY number { HOUR | HOURS | DAY | DAYS | WEEK | WEEKS } |
  CRON cron_string [ AT TIME ZONE timezone_id ] }

Parameter

  • REFRESH

    Bei Angabe dieser Option wird die Tabelle mit den neuesten Daten aktualisiert, die aus den in der Abfrage definierten Quellen verfügbar sind. Nur neue Daten, die vor dem Start der Abfrage eingehen, werden verarbeitet. Neue Daten, die den Quellen während der Ausführung des Befehls hinzugefügt werden, werden bis zur nächsten Aktualisierung ignoriert. Der Aktualisierungsvorgang aus CREATE OR REFRESH ist vollständig deklarativ. Wenn ein Aktualisierungsbefehl nicht alle Metadaten aus der ursprünglichen Tabellenerstellungsanweisung angibt, werden die nicht angegebenen Metadaten gelöscht.

  • IF NOT EXISTS

    Erstellt die Streamingtabelle, sofern sie noch nicht vorhanden ist. Wenn bereits eine Tabelle mit diesem Namen vorhanden ist, wird die CREATE STREAMING TABLE-Anweisung ignoriert.

    Sie können nur eines von IF NOT EXISTS oder OR REFRESH angeben.

  • table_name

    Der Name der zu erstellenden Tabelle. Der Name darf keine zeitliche Spezifikation oder Optionsspezifikation enthalten. Wenn der Name ohne Qualifizierung angegeben wird, wird die Tabelle im aktuellen Schema erstellt.

  • table_specification

    Diese optionale Klausel definiert die Liste der Spalten, deren Typen, Eigenschaften, Beschreibungen und Spalteneinschränkungen.

    Wenn Sie keine Spalten im Tabellenschema definieren, müssen Sie AS query angeben.

    • column_identifier

      Ein eindeutiger Name für die Spalte.

      • column_type

        Gibt den Datentyp der Spalte an.

      • NOT NULL

        Bei Angabe dieser Option akzeptiert die Spalte keine NULL-Werte.

      • COMMENT column_comment

        Ein Zeichenfolgenliteral zum Beschreiben der Spalte.

      • column_constraint

        Wichtig

        Dieses Feature befindet sich in der Public Preview.

        Fügt der Spalte in einer Streamingtabelle eine Primärschlüssel- oder Fremdschlüsseleinschränkung hinzu. Einschränkungen werden nicht für Tabellen im hive_metastore-Katalog unterstützt.

      • MASK-Klausel

        Wichtig

        Dieses Feature befindet sich in der Public Preview.

        Fügt eine Spaltenmaskierungsfunktion hinzu, um vertrauliche Zeichenfolgenwerte zu anonymisieren. Alle nachfolgenden Abfragen über diese Spalte erhalten das Ergebnis der Auswertung dieser Funktion über die Spalte anstelle des ursprünglichen Werts der Spalte. Dies kann für eine präzise Zugriffssteuerung nützlich sein, bei der die Funktion die Identität oder Gruppenmitgliedschaft der aufrufenden Benutzenden überprüfen kann, um zu entscheiden, ob der Wert zurückgezogen werden soll.

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

        Fügt der Tabelle Datenqualitätserwartungen hinzu. Diese Datenqualitätserwartungen können im Laufe der Zeit nachverfolgt werden, und der Zugriff darauf ist über das Ereignisprotokoll der Streamingtabelle möglich. 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.

        expectation_expr kann aus Literalen, Spaltenbezeichnern innerhalb der Tabelle und deterministischen, integrierten SQL-Funktionen oder -Operatoren bestehen, mit Ausnahme von:

        Darüber hinaus darf expr keine Unterabfrage enthalten.

      • table_constraint

        Wichtig

        Dieses Feature befindet sich in der Public Preview.

        Fügt einer Streamingtabelle eine informative Primärschlüssel- oder Fremdschlüsseleinschränkung hinzu. Schlüsseleinschränkungen werden nicht für Tabellen im hive_metastore-Katalog unterstützt.

  • table_clauses

    Geben Sie optional Partitionierung, Kommentare, benutzerdefinierte Eigenschaften und einen Aktualisierungszeitplan für die neue Tabelle an. Jede Unterklausel kann nur einmal angegeben werden.

    • PARTITIONED BY

      Eine optionale Liste der Spalten der Tabelle, nach denen die Tabelle partitioniert werden soll.

      Hinweis

      Flüssigkeitsclustering bietet eine flexible, optimierte Lösung für Clustering. Erwägen Sie die Verwendung von CLUSTER BY anstelle von PARTITIONED BY für Streamingtabellen.

    • CLUSTER BY

      Eine optionale Klausel zum Gruppieren nach einer Teilmenge von Spalten. Weitere Informationen zum Flüssigclustering finden Sie unter Verwenden von Flüssigclustering für Delta-Tabellen.

      Delta Lake Liquid Clustering kann nicht mit PARTITIONED BY kombiniert werden.

    • COMMENT table_comment

      Ein STRING-Literal zum Beschreiben der Tabelle.

    • TBLPROPERTIES

      Legt optional eine oder mehrere benutzerdefinierte Eigenschaften fest.

      Verwenden Sie diese Einstellung, um den Delta Live Tables-Laufzeitkanal anzugeben, der zum Ausführen dieser Anweisung verwendet wird. Legen Sie den Wert der pipelines.channel-Eigenschaft auf "PREVIEW" oder "CURRENT" fest. Standardwert: "CURRENT". Weitere Informationen zu Delta Live Tables-Kanälen finden Sie unter Delta Live Tables-Laufzeitkanäle.

    • ZEITPLAN [ REFRESH ] Planungsklausel

    • EVERY number { HOUR | HOURS | DAY | DAYS | WEEK | WEEKS }

      Verwenden Sie die EVERY-Syntax, um eine Aktualisierung zu planen, die in regelmäßigen Abständen auftritt. Wenn die EVERY-Syntax angegeben ist, wird die Streamingtabelle oder materialisierte Sicht basierend auf dem angegebenen Wert (z. B. HOUR, HOURS, DAY, DAYS, WEEK oder WEEKS) regelmäßig im angegebenen Intervall aktualisiert. In der folgenden Tabelle finden Sie zulässige ganzzahlige Werte für number.

      Zeiteinheit Ganzzahliger Wert
      HOUR or HOURS 1 <= H <= 72
      DAY or DAYS 1 <= D <= 31
      WEEK or WEEKS 1 <= W <= 8

      Hinweis

      Die Singular- und Pluralformen der enthaltenen Zeiteinheit sind semantisch gleichwertig.

    • CRON cron_string [ AT TIME ZONE timezone_id ]

      Zum Planen einer Aktualisierung mithilfe eines quartz cron-Werts. Gültige time_zone_values-Werte werden akzeptiert. AT TIME ZONE LOCAL wird nicht unterstützt.

      Fehlt AT TIME ZONE, wird die Sitzungszeitzone verwendet. Wenn AT TIME ZONE fehlt und die Sitzungszeitzone nicht festgelegt ist, wird ein Fehler ausgelöst. SCHEDULE ist semantisch äquivalent mit SCHEDULE REFRESH.

    Der Zeitplan kann als Teil des CREATE-Befehls bereitgestellt werden. Verwenden Sie ALTER STREAMING TABLE oder führen Sie CREATE OR REFRESH Befehl mit SCHEDULE Klausel aus, um den Zeitplan einer Streamingtabelle nach der Erstellung zu ändern.

  • WITH ROW FILTER-Klausel

    Wichtig

    Dieses Feature befindet sich in der Public Preview.

    Fügt der Tabelle eine Zeilenfilterfunktion hinzu. Alle nachfolgenden Abfragen über diese Tabelle erhalten eine Teilmenge der Zeilen, wo die Funktion als boolescher TRUE-Wert ausgewertet wird. Dies kann für eine präzise Zugriffssteuerung nützlich sein, bei der die Funktion die Identität oder Gruppenmitgliedschaft der aufrufenden Benutzenden überprüfen kann, um zu entscheiden, ob bestimmte Zeilen gefiltert werden sollen.

  • AS query

    Diese Klausel füllt die Tabelle mit den Daten aus query auf. Diese Abfrage muss eine Streamingabfrage sein. Dies kann erreicht werden, indem Sie jeder Beziehung, die Sie inkrementell verarbeiten möchten, das Schlüsselwort STREAM hinzufügen. Bei gemeinsamer Angabe von query und table_specification muss das in table_specification angegebene Tabellenschema alle von query zurückgegebenen Spalten enthalten. Andernfalls tritt ein Fehler auf. Alle Spalten, die in table_specification angegeben, aber nicht von query zurückgegeben wurden, geben bei der Abfrage null-Werte zurück.

Unterschiede zwischen Streamingtabellen und anderen Tabellen

Streamingtabellen sind zustandsbehaftete Tabellen, die so konzipiert sind, dass jede Zeile nur einmal verarbeitet wird, während Sie ein wachsendes Dataset verarbeiten. Da die meisten Datasets im Laufe der Zeit kontinuierlich wachsen, eignen sich Streamingtabellen für die meisten Erfassungsworkloads. Streamingtabellen sind optimal für Pipelines, die Datenfrische und geringe Wartezeiten erfordern. Streamingtabellen können auch für Transformationen im großen Stil nützlich sein, da die Ergebnisse beim Eintreffen neuer Daten inkrementell berechnet werden können, um die Ergebnisse auf dem neuesten Stand zu halten, ohne dass alle Quelldaten mit jedem Update vollständig neu verarbeitet werden müssen. Streamingtabellen sind für Datenquellen mit reinem Anfügen konzipiert.

Streamingtabellen akzeptieren zusätzliche Befehle wie REFRESH. Mit diesem Befehl werden die neuesten Daten verarbeitet, die in den in der Abfrage angegebenen Quellen verfügbar sind. Änderungen an der angegebenen Abfrage werden nur in neuen Daten berücksichtigt, indem REFRESH aufgerufen wird, aber nicht in zuvor verarbeiteten Daten. Sollen die Änderungen auch auf vorhandene Daten angewendet werden, müssen Sie REFRESH TABLE <table_name> FULL ausführen, um eine vollständige Aktualisierung (FULL REFRESH) zu erreichen. Vollständige Aktualisierungen verarbeiten alle in der Quelle verfügbaren Daten mit der neuesten Definition erneut. Es wird nicht empfohlen, vollständige Aktualisierungen für Quellen aufzurufen, die nicht den gesamten Datenverlauf beibehalten oder kurze Aufbewahrungszeiträume aufweisen (z. B. Kafka), da durch eine vollständige Aktualisierung die vorhandenen Daten abgeschnitten werden. Möglicherweise können Sie alte Daten nicht wiederherstellen, wenn die Daten in der Quelle nicht mehr verfügbar sind.

Zeilenfilter und Spaltenmasken

Wichtig

Dieses Feature befindet sich in der Public Preview.

Mit Zeilenfiltern können Sie eine Funktion angeben, die als Filter gilt, wenn ein Tabellenscan Zeilen abruft. Mit diesen Filtern können Sie sicherstellen, dass nachfolgende Abfragen nur Zeilen zurückgeben, für die das Filterprädikat zu TRUE ausgewertet wird.

Mit Spaltenmasken können Sie die Werte einer Spalte immer filtern, wenn ein Tabellenscan Zeilen abruft. Alle zukünftigen Abfragen dieser Spalte erhalten das Ergebnis der Auswertung dieser Funktion für die Spalte, das den ursprünglichen Spaltenwert ersetzt.

Weitere Informationen zur Verwendung von Zeilenfiltern und Spaltenmasken finden Sie unter Filtern vertraulicher Tabellendaten mit Zeilenfiltern und Spaltenmasken.

Verwalten von Zeilenfiltern und Spaltenmasken

Zeilenfilter und Spaltenmasken für Streamingtabellen sollten der CREATE OR REFRESH -Anweisung hinzugefügt, aktualisiert oder gelöscht werden.

Verhalten

  • Als Definierer aktualisieren: Wenn die CREATE OR REFRESH- oder REFRESH-Anweisungen eine Streamingtabelle aktualisieren, werden Zeilenfilterfunktionen mit den Berechtigungen des Definierers (als Tabellenbesitzer) ausgeführt. Dies bedeutet, dass die Tabellenaktualisierung den Sicherheitskontext des Benutzers verwendet, der die Streamingtabelle erstellt hat.
  • Abfrage: Während die meisten Filter mit den Berechtigungen des Definierers ausgeführt werden, sind Funktionen wie CURRENT_USER und IS_MEMBER, die den Benutzerkontext überprüfen, die Ausnahme. Diese Funktionen werden als Aufrufer ausgeführt. Durch diese Vorgehensweise werden benutzerspezifische Datensicherheit und Zugriffssteuerungen basierend auf dem Kontext der aktuellen benutzenden Person erzwungen.

Einblick

Verwenden Sie DESCRIBE EXTENDED, INFORMATION_SCHEMA oder den Katalog-Explorer, um die vorhandenen Zeilenfilter und Spaltenmasken zu untersuchen, die für eine bestimmte Streamingtabelle gelten. Mit dieser Funktionalität können Benutzende Datenzugriffs- und Schutzmaßnahmen für Streamingtabellen überwachen und überprüfen.

Einschränkungen

  • Nur Tabellenbesitzende können Streamingtabellen aktualisieren, um die neuesten Daten abzurufen.
  • ALTER TABLE-Befehle sind für Streamingtabellen nicht zulässig. Die Definition und die Eigenschaften der Tabelle sollten durch die CREATE OR REFRESH- oder ALTER STREAMING TABLE-Anweisung geändert werden.
  • Die Weiterentwicklung des Tabellenschemas durch DML-Befehle wie INSERT INTO und MERGE wird nicht unterstützt.
  • Die folgenden Befehle werden für Streamingtabellen nicht unterstützt:
    • CREATE TABLE ... CLONE <streaming_table>
    • COPY INTO
    • ANALYZE TABLE
    • RESTORE
    • TRUNCATE
    • GENERATE MANIFEST
    • [CREATE OR] REPLACE TABLE
  • Delta Sharing wird nicht unterstützt.
  • Das Umbenennen der Tabelle oder das Ändern des Besitzers bzw. der Besitzerin wird nicht unterstützt.
  • Tabelleneinschränkungen wie PRIMARY KEY und FOREIGN KEY werden nicht unterstützt.
  • Generierte Spalten, Identitätsspalten und Standardspalten werden nicht unterstützt.

Beispiele

-- Creates a streaming table that processes files stored in the given external location with
-- schema inference and evolution.
> CREATE OR REFRESH STREAMING TABLE raw_data
  AS SELECT * FROM STREAM read_files('abfss://container@storageAccount.dfs.core.windows.net/base/path');

-- Creates a streaming table that processes files with a known schema.
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int,
    ts timestamp,
    event string
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');

-- Creates a streaming table with schema evolution and data quality expectations.
-- The table creation or refresh fails if the data doesn't satisfy the expectation.
> CREATE OR REFRESH STREAMING TABLE avro_data (
    CONSTRAINT date_parsing EXPECT (to_date(dt) >= '2000-01-01') ON VIOLATION FAIL UPDATE
  )
  AS SELECT *
  FROM STREAM read_files('gs://my-bucket/avroData');

-- Stores the data from Kafka in an append-only streaming table.
> CREATE OR REFRESH STREAMING TABLE firehose_raw
  COMMENT 'Stores the raw data from Kafka'
  TBLPROPERTIES ('delta.appendOnly' = 'true')
  AS SELECT
    value raw_data,
    offset,
    timestamp,
    timestampType
  FROM STREAM read_kafka(bootstrapServers => 'ips', subscribe => 'topic_name');

-- Sets the runtime channel to "PREVIEW"
> CREATE STREAMING TABLE st_preview
  TBLPROPERTIES(pipelines.channel = "PREVIEW")
  AS SELECT * FROM RANGE(10)

-- Read data from another streaming table scheduled to run every hour.
> CREATE STREAMING TABLE firehose_bronze
  SCHEDULE EVERY 1 HOUR
  AS SELECT
    from_json(raw_data, 'schema_string') data,
    * EXCEPT (raw_data)
  FROM STREAM firehose_raw;

-- Creates a streaming table with a column constraint
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int PRIMARY KEY,
    ts timestamp,
    event string
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');

-- Creates a streaming table with a table constraint
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int,
    ts timestamp,
    event string,
    CONSTRAINT pk_id PRIMARY KEY (id)
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');

-- Creates a streaming table with a row filter and a column mask
> CREATE OR REFRESH STREAMING TABLE masked_csv_data (
    id int,
    name string,
    region string,
    ssn string MASK catalog.schema.ssn_mask_fn
  )
  WITH ROW FILTER catalog.schema.us_filter_fn ON (region)
  AS SELECT *
  FROM STREAM read_files('s3://bucket/path/sensitive_data')