CREATE STREAMING TABLE

Gilt für: Databricks SQL Databricks Runtime ab Version 13.3 LTS

Wichtig

Dieses Feature befindet sich in der Public Preview.

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. Weitere Informationen finden Sie unter Implementieren einer Delta Live Tables-Pipeline 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 [ NOT NULL ]
      [ COMMENT column_comment ] [ column_constraint ]
    ] [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
      [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...]
    [ , table_constraint ] [...] )

table_clauses
  { PARTITIONED BY (col [, ...]) |
    COMMENT table_comment |
    TBLPROPERTIES clause |
    SCHEDULE [ REFRESH ] 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.

• IF NOT EXISTS

  Wenn dies angegeben wird und bereits eine Tabelle gleichen Namens vorhanden ist, wird die Anweisung ignoriert.

  IF NOT EXISTS kann nicht zusammen mit REFRESH verwendet werden, was bedeutet, dass CREATE OR REFRESH TABLE IF NOT EXISTS nicht zulässig ist.

• table_name

  Der Name der zu erstellenden Tabelle. Der Name darf keine temporale Spezifikation 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.

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

      • Aggregatfunktionen
        • Analysefensterfunktionen
        • Fensterrangfunktionen
        • Tabellenwertgenerator-Funktionen

      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.

  • COMMENT table_comment

    Ein STRING-Literal zum Beschreiben der Tabelle.

  • TBLPROPERTIES

    Legt optional eine oder mehrere benutzerdefinierte Eigenschaften fest.

  • SCHEDULE [ REFRESH ] CRON cron_string [ AT TIME ZONE timezone_id ]

    Falls angegeben, legt dieser Parameter fest, dass die Daten der Streamingtabelle oder der materialisierten Sicht nach dem angegebenen quarz cron-Zeitplan aktualisiert werden. Es werden nur time_zone_values 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.

    Sie können die Syntax SCHEDULE nicht in einer Delta Live Tables-Pipelinedefinition verwenden.

    Die SCHEDULE-Klausel ist in einem CREATE OR REFRESH-Befehl nicht zulässig. Der Zeitplan kann als Teil des CREATE-Befehls bereitgestellt werden. Verwenden Sie ALTER STREAMING TABLE, um den Zeitplan einer Streamingtabelle nach der Erstellung zu ändern.

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

  Diese Klausel ist für Streamingtabellen erforderlich, die in Databricks SQL erstellt wurden. In Delta Live Tables ist sie dagegen nicht erforderlich. Wenn diese Klausel in Delta Live Tables nicht angegeben wird, müssen Sie auf diese Tabelle in einem APPLY CHANGES-Befehl in Ihrer DLT-Pipeline verweisen. Weitere Informationen finden Sie unter Change data capture with SQL in Delta Live Tables.

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.

Begrenzungen

• Nur Tabellenbesitzer*innen 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 über die ALTER STREAMING TABLE-Anweisung geändert werden.

• Zeitreiseabfragen werden nicht unterstützt.

• Die Weiterentwicklung des Tabellenschemas durch DML-Befehle wie INSERT INTO und MERGE wird nicht unterstützt.

• Folgende 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 (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');

-- Read data from another streaming table scheduled to run every hour.
> CREATE STREAMING TABLE firehose_bronze
  SCHEDULE CRON '0 0 * * * ? *'
  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');

Verwandte Artikel

• read_files

• read_kafka

• event_log

• DROP TABLE

• Tabelleneigenschaften und Tabellenoptionen

• CONSTRAINT

• PARTITIONED BY

• CREATE MATERIALIZED VIEW

• Informationsschema

• REFRESH TABLE

• ALTER STREAMING TABLE