Streamingtabellen

Eine Streamingtabelle ist eine Delta-Tabelle mit zusätzlicher Unterstützung für Streaming oder inkrementelle Datenverarbeitung. Eine Streamingtabelle kann durch einen oder mehrere Datenströme in einer ETL-Pipeline angesprochen werden.

Streamingtabellen sind aus folgenden Gründen eine gute Wahl für die Datenaufnahme:

• Jede Eingabezeile wird nur einmal behandelt, wodurch die überwiegende Mehrheit der Erfassungsworkloads modelliert wird (d. a. indem Zeilen an eine Tabelle angefügt oder heraufgestrigt werden).
• Sie können große Mengen von Daten verarbeiten, die nur hinzugefügt werden können.

Streamingtabellen sind aus folgenden Gründen auch eine gute Wahl für Streamingtransformationen mit geringer Latenz:

• Grund für Zeilen und Zeitfenster
• Große Datenmengen verarbeiten
• Geringe Latenz

Das folgende Diagramm veranschaulicht, wie Streamingtabellen funktionieren.

Bei jeder Aktualisierung lesen die Flüsse, die einer Streamingtabelle zugeordnet sind, die geänderten Informationen in einer Streamingquelle und fügen neue Informationen an diese Tabelle an.

Streamingtabellen werden von einer einzelnen Pipeline definiert und aktualisiert. Sie definieren Streamingtabellen explizit im Quellcode der Pipeline. Tabellen, die von einer Pipeline definiert sind, können von keiner anderen Pipeline geändert oder aktualisiert werden. Sie können mehrere Flüsse definieren, die an eine einzelne Streamingtabelle angefügt werden sollen.

Wenn Sie eine Streamingtabelle außerhalb einer Pipeline in Databricks SQL erstellen, erstellt Databricks eine ausgeblendete Pipeline, die zum Aktualisieren dieser Tabelle verwendet wird.

Weitere Informationen zu Flüssen finden Sie unter Daten inkrementell laden und verarbeiten mit Lakeflow Declarative Pipelines Flüssen.

Streamingtabellen für die Aufnahme

Streamingtabellen sind nur für Anfügedatenquellen vorgesehen und verarbeiten Eingaben nur einmal.

Das folgende Beispiel zeigt, wie Sie eine Streamingtabelle verwenden, um neue Dateien aus dem Cloudspeicher aufzunehmen.

Python

import dlt

# create a streaming table
@dlt.table
def customers_bronze():
  return (
    spark.readStream.format("cloudFiles")
     .option("cloudFiles.format", "json")
     .option("cloudFiles.inferColumnTypes", "true")
     .load("/Volumes/path/to/files")
  )

Wenn Sie die spark.readStream Funktion in einer Datasetdefinition verwenden, bewirkt dies, dass Lakeflow Declarative Pipelines das Dataset als Datenstrom behandeln, und die erstellte Tabelle ist eine Streamingtabelle.

SQL

-- create a streaming table
CREATE OR REFRESH STREAMING TABLE customers_bronze
AS SELECT * FROM STREAM read_files(
  "/volumes/path/to/files",
  format => "json"
);

Weitere Informationen zum Laden von Daten in die Streamingtabelle finden Sie unter "Laden von Daten mit Lakeflow Declarative Pipelines".

Das folgende Diagramm veranschaulicht, wie Nur-Anfüge-Streamingtabellen funktionieren.

Eine Zeile, die bereits an eine Streamingtabelle angefügt wurde, wird nicht mit späteren Aktualisierungen der Pipeline erneut abgefragt. Wenn Sie die Abfrage ändern (z. B. von SELECT LOWER (name) zu SELECT UPPER (name)), werden vorhandene Zeilen nicht in Großbuchstaben aktualisiert, aber neue Zeilen sind Großbuchstaben. Sie können eine vollständige Aktualisierung auslösen, um alle vorherigen Daten aus der Quelltabelle neu abzufragen. Dadurch werden alle Zeilen in der Streamingtabelle aktualisiert.

Streamingtabellen und Streaming mit geringer Latenz

Streamingtabellen sind für Streaming mit geringer Latenz über den gebundenen Zustand ausgelegt. Streamingtabellen verwenden die Prüfpunktverwaltung, wodurch sie gut für Streaming mit geringer Latenz geeignet sind. Sie erwarten jedoch Ströme, die natürlich gebunden oder mit einem Wasserzeichen begrenzt sind.

Ein natürlich gebundener Datenstrom wird von einer Streamingdatenquelle erzeugt, die einen klar definierten Start- und End-Vorgang aufweist. Ein Beispiel für einen natürlich gebundenen Datenstrom ist das Lesen von Daten aus einem Verzeichnis mit Dateien, in denen keine neuen Dateien hinzugefügt werden, nachdem ein anfänglicher Batch von Dateien platziert wurde. Der Datenstrom wird als gebunden betrachtet, da die Anzahl der Dateien endlich ist und dann der Datenstrom endet, nachdem alle Dateien verarbeitet wurden.

Sie können auch ein Wasserzeichen verwenden, um einen Strom zu binden. Ein Wasserzeichen in Spark Structured Streaming ist ein Mechanismus, mit dem späte Daten verarbeitet werden können, indem angegeben wird, wie lange das System auf verzögerte Ereignisse warten soll, bevor das Zeitfenster als abgeschlossen betrachtet wird. Ein ungebundener Datenstrom ohne Wasserzeichen kann dazu führen, dass eine Pipeline aufgrund von Speicherdruck fehlschlägt.

Weitere Informationen zur zustandsbehafteten Datenstromverarbeitung finden Sie unter Optimieren der zustandsbehafteten Verarbeitung in Lakeflow Declarative Pipelines mit Wasserzeichen.For more information about stateful stream processing, see Optimize stateful processing in Lakeflow Declarative Pipelines with watermarks.

Stream-Snapshot-Verknüpfungen

Stream-Snapshot-Verknüpfungen sind Verknüpfungen zwischen einem Datenstrom und einer Dimension, von der beim Start von Datenströmen eine Momentaufnahme erstellt wird. Diese Verknüpfungen werden nicht neu kompensiert, wenn sich die Dimension nach dem Start des Datenstroms ändert, da die Dimensionstabelle als Momentaufnahme behandelt wird und Änderungen an der Dimensionstabelle nach dem Start des Datenstroms nicht wiedergegeben werden, es sei denn, Sie laden die Dimensionstabelle neu oder aktualisieren sie. Dies ist ein vernünftiges Verhalten, wenn Sie kleine Abweichungen in einer Verknüpfung akzeptieren können. Beispielsweise ist eine ungefähre Verknüpfung akzeptabel, wenn die Anzahl der Transaktionen viele Größenordnungen größer als die Anzahl der Kunden ist.

Im folgenden Codebeispiel verknüpfen wir eine Dimensionstabelle, Kunden mit zwei Zeilen mit einem immer größer werdenden Dataset, Transaktionen. Wir materialisieren eine Verknüpfung zwischen diesen beiden Datasets in einer Tabelle mit dem Namen .We materialize a join between these two datasets in a table called sales_report. Beachten Sie, dass, wenn ein externer Prozess die Kundentabelle aktualisiert, indem eine neue Zeile (customer_id=3, name=Zoya) hinzugefügt wird, diese neue Zeile NICHT in der Verknüpfung vorhanden ist, weil die statische Dimensionstabelle zu dem Zeitpunkt aufgenommen wurde, als die Datenströme gestartet wurden.

import dlt

@dlt.view
# assume this table contains an append-only stream of rows about transactions
# (customer_id=1, value=100)
# (customer_id=2, value=150)
# (customer_id=3, value=299)
# ... <and so on> ...
def v_transactions():
  return spark.readStream.table("transactions")

# assume this table contains only these two rows about customers
# (customer_id=1, name=Bilal)
# (customer_id=2, name=Olga)
@dlt.view
def v_customers():
  return spark.read.table("customers")

@dlt.table
def sales_report():
  facts = spark.readStream.table("v_transactions")
  dims = spark.read.table("v_customers")

  return (
    facts.join(dims, on="customer_id", how="inner"
  )

Einschränkungen der Streamingtabelle

Streamingtabellen haben die folgenden Einschränkungen:

• Begrenzte Entwicklung: Sie können die Abfrage ändern, ohne das gesamte Dataset neu zu komputieren. Da eine Streamingtabelle nur einmal eine Zeile sieht, können Sie unterschiedliche Abfragen für unterschiedliche Zeilen verwenden. Dies bedeutet, dass Sie alle vorherigen Versionen der Abfrage kennen müssen, die in Ihrem Dataset ausgeführt werden. Eine vollständige Aktualisierung ist erforderlich, um die Daten der Streamingtabelle zu aktualisieren, die bereits verarbeitet wurden.
• Zustandsverwaltung: Streamingtabellen haben eine geringe Latenz, sodass Sie sicherstellen müssen, dass die Datenströme, die sie betreiben, natürlich gebunden oder mit Wasserzeichen gebunden sind. Weitere Informationen finden Sie unter Optimierung der zustandsbehafteten Verarbeitung in Lakeflow Declarative Pipelines mit Wasserzeichen.
• Verknüpfungen werden nicht neu berechnet: Verknüpfungen in Streamingtabellen werden nicht neu berechnet, wenn sich Dimensionen ändern. Diese Eigenschaft kann für "schnelle, aber falsche" Szenarien gut sein. Wenn Ihre Ansicht immer korrekt sein soll, können Sie eine materialisierte Ansicht verwenden. Materialisierte Ansichten sind immer richtig, da sie Verknüpfungen automatisch neu kompensieren, wenn sich Dimensionen ändern. Weitere Informationen finden Sie unter Materialisierte Ansichten.