Kostengünstige Daten in Echtzeit mit dbt und Databricks liefern

Wenn Unternehmen wachsen, skalieren die Datenmengen von GB auf TB (oder mehr) und die Latenzanforderungen sinken von Stunden auf Minuten (oder weniger), was es immer teurer macht, dem Unternehmen frische Erkenntnisse zu liefern. Historisch gesehen haben Python- und Scala-Dateningenieure auf Streaming zurückgegriffen, um diese Anforderungen zu erfüllen und neue Daten in Echtzeit effizient zu verarbeiten. Analytik-Ingenieure, die SQL-basierte dbt-Pipelines skalieren mussten, hatten diese Option jedoch nicht.

Nicht mehr! Dieser Blogbeitrag soll veranschaulichen, wie wir die neuen Streaming Tables und Materialized Views auf Databricks nutzen können, um mit der Einfachheit von SQL und dbt frische, Echtzeit-Erkenntnisse für Unternehmen zu liefern.

Hintergrund

Auf dem Data + AI Summit 2023 haben wir Streaming Tables und Materialized Views in Databricks SQL eingeführt. Diese großartige Funktion gab Databricks SQL-Benutzern einfachen Zugriff auf leistungsstarke neue Tabellenmaterialisierungen, die zuerst in Delta Live Tables eingeführt wurden. Dies ermöglichte ihnen, große Abfragen inkrementell zu verarbeiten, direkt aus Event-Datenquellen zu streamen und mehr.

Neben der nativen Nutzung von Streaming Tables und Materialized Views innerhalb einer Databricks-Umgebung funktionieren sie auch für dbt-Benutzer auf Databricks. dbt-databricks ist zu einer der beliebtesten Methoden geworden, um Datenmodelle auf Databricks zu erstellen. Dabei werden alle leistungsstarken Funktionen von Databricks SQL genutzt, einschließlich der Photon Compute Engine, der sofort skalierbaren Serverless SQL Warehouses und des Unity Catalog Governance-Modells, mit der Allgegenwart des Transformations-Frameworks von dbt.

Was hat sich in dbt-databricks geändert?

Ab dbt v1.6+ hat sich dbt-databricks in drei Schlüsselaspekten weiterentwickelt:

1. Neue Materialisierungen: "streaming_table" und "materialized_view"
2. Neue Syntax zum direkten Lesen aus Cloud-Datenspeichern, ohne die Quellen als Tabelle zu stagen
3. Zugriff auf erweiterte Streaming-Konzepte wie Window Aggregations, Watermarking und Stream-Stream Joins

Hinweis: Halten Sie Ausschau nach der bevorstehenden dbt v1.7.3-Version, die die oben genannten Funktionen weiter verfeinern wird!

Werfen wir einen Blick darauf, wie wir diese neuen Funktionen mit der Airline Trips Demo nutzen können.

Die Airline Trips Demo

Die Airline Trips Demo wurde entwickelt, um zu demonstrieren, wie Live-Event-Daten inkrementell aufgenommen und transformiert werden können, um aktuelle Geschäftseinblicke auf Databricks zu erhalten, sei es für ein Dashboard oder ein KI-Modell. Der Datensatz repräsentiert alle Flugreisen in den Vereinigten Staaten im Laufe der Zeit und erfasst die Verspätungen bei Abflügen und Ankünften für jede Reise.

Ein enthaltenes Hilfs-Notebook richtet einen simulierten Stream aus diesem Datensatz ein, während das dbt-Projekt ein Datenmodell zeigt, das diese rohen JSON-Events aufnimmt und sie über Streaming ETL in eine Schicht von Materialized Views, Feature Tables und mehr transformiert.

Das Repository ist hier öffentlich verfügbar und nutzt Beispieldaten, die in allen Databricks Workspaces standardmäßig enthalten sind. Fühlen Sie sich frei, mitzumachen!

Daten aus Cloud-Datenspeichern aufnehmen

Eine der einfachsten Möglichkeiten, Streaming Tables zu nutzen, ist die Datenaufnahme aus Cloud-Datenspeichern wie S3 für AWS oder ADLS für Azure. Möglicherweise haben Sie eine vorgelagerte Datenquelle, die Event-Daten in hohem Volumen generiert, und einen Prozess, um diese als Rohdateien an einem Speicherort abzulegen, typischerweise im JSON-, CSV-, Parquet- oder Avro-Format.

In unserer Demo stellen wir uns vor, wir erhalten einen Live-Feed jeder Flugreise in den Vereinigten Staaten von einem externen Anbieter und möchten diese inkrementell aufnehmen, sobald sie eintreffen.

Anstatt die Dateien als externe Tabelle zu stagen oder ein Drittanbieter-Tool zur Materialisierung einer Delta-Tabelle für die Datenquelle zu verwenden, können wir einfach Streaming Tables verwenden, um dies zu lösen. Betrachten Sie das folgende Modell für unseren Bronze-Airline-Trips-Feed:

Die beiden wichtigsten Punkte, die zu beachten sind:

• Die Materialisierungsstrategie ist auf 'streaming_table' gesetzt
  • Dies führt einen CREATE OR REFRESH STREAMING TABLE Befehl in Databricks aus
• Die Syntax zum Lesen aus Cloud-Speichern nutzt Auto Loader im Hintergrund
  • read_files() listet neue JSON-Dateien im angegebenen Ordner auf und beginnt mit der Verarbeitung. Da wir dbt verwenden, haben wir die var()-Funktion in dbt genutzt, um dynamisch einen S3-Ordnerpfad zu übergeben (in der Form "s3://…")
  • Das STREAM-Schlüsselwort gibt an, von diesem Speicherort zu streamen. Alternativ können wir ohne dieses auch read_files() mit materialized='table' verwenden, um einen Batch-Lesevorgang direkt aus dem angegebenen Ordner durchzuführen

Nebenbei bemerkt: Während Auto Loader die geringste Einrichtung erfordert, können Sie auch direkt von einer Event-Streaming-Plattform wie Kafka, Kinesis oder Event Hubs für noch geringere Latenzzeiten mit sehr ähnlicher Syntax streamen. Weitere Details finden Sie hier.

Daten inkrementell für die Silver-Schicht anreichern

Streaming muss nicht am Ingestionsschritt enden. Wenn wir nachgelagert einige Joins durchführen oder einen Surrogat-Schlüssel hinzufügen möchten, aber nur für neue Daten, um Rechenleistung zu sparen, können wir weiterhin die Streaming Table-Materialisierung verwenden. Nehmen Sie zum Beispiel den Ausschnitt aus unserem nächsten Modell für die Silver-Schicht, die angereicherten Airline-Trips-Daten, bei denen wir Zuordnungstabellen für Flughafencodes in den Rohdatensatz einfügen:

Auch hier haben wir die Streaming Table-Materialisierung genutzt und konnten für unsere gesamte Logik Standard-dbt-Funktionalität verwenden. Dazu gehören:

• Nutzung des dbt_utils-Pakets für praktische Abkürzungen wie die Generierung eines Surrogat-Schlüssels
• Verwendung der ref()-Anweisung zur Aufrechterhaltung der vollständigen Datenherkunft (Lineage)

Die einzige wirkliche Änderung an unserem SQL war die Hinzufügung des STREAM()-Schlüsselworts um die ref()-Anweisung für airline_trips_bronze, um anzuzeigen, dass diese Tabelle inkrementell gelesen wird, während die verknüpfte airport_codes-Tabelle eine Mapping-Tabelle ist, die vollständig gelesen wird. Dies wird als Stream-Static Join bezeichnet.

Erstellung einer recheneffizienten Gold-Schicht mit Materialized Views

Nachdem unsere angereicherten Silver-Tabellen bereit sind, können wir darüber nachdenken, wie wir aggregierte Erkenntnisse für unsere Endbenutzer bereitstellen möchten. Wenn wir normalerweise eine Tabellenmaterialisierung verwenden würden, müssten wir jedes Mal alle historischen Ergebnisse neu berechnen.

Um die Vorteile der vorgelagerten Streaming Tables zu nutzen, die nur neue Daten in jedem Durchlauf verarbeiten, greifen wir stattdessen auf Materialized Views für diese Aufgabe zurück!

Die gute Nachricht in Databricks ist, dass ein Modell, das eine Materialized View erstellt, nicht anders aussieht als ein Modell, das eine Tabelle erstellt! Betrachten Sie unser Beispiel für eine Materialized View der Gold-Schicht zur Berechnung des Prozentsatzes verspäteter Flüge pro Tag:

Wir haben nur die Materialisierungs-Konfiguration geändert!

Denken Sie daran, dass Materialized Views inkrementell aktualisiert werden können, wenn es Änderungen an den Basistabellen gibt. Im obigen Szenario bestimmt die Materialized View, welche Gruppen neu berechnet werden müssen, und berechnet nur diese, während unveränderte Aggregationen unverändert bleiben und die Gesamtrechenkosten gesenkt werden. Dies ist im Beispiel leichter zu visualisieren, da wir über ArrDate, das Ankunftsdatum der Flüge, aggregieren. Das bedeutet, dass neue Tagesdaten natürlich in neue Gruppen fallen und bestehende Gruppen unverändert bleiben.

Wenn wir die Ereignisprotokolle der Materialized View (unten abgebildet) nach mehreren Durchläufen des Modells analysieren, sehen wir die inkrementelle Verarbeitung am Werk. Der erste Durchlauf ist eine vollständige Berechnung wie bei jeder Tabelle, aber ein zweiter Durchlauf zur Aktualisierung der Aggregationen mit neuen Daten nutzt eine zeilenweise inkrementelle Aktualisierung. Ein letzter Durchlauf des Modells erkannte, dass keine neuen Daten vorgelagert aufgenommen wurden, und tat einfach nichts.

Was gibt es noch im Demo-Repository?

Wir haben die Grundlagen abgedeckt, um Daten direkt aus der Ereignisquelle in eine BI-taugliche Materialized View zu bekommen, aber das Demo-Repository enthält noch viel mehr.

Das Repository enthält Beispiele, wie Sie Protokolle für Streaming Tables und Materialized Views überwachen können, um zu verstehen, wie Daten verarbeitet werden, sowie ein fortgeschrittenes Beispiel, das in diesem Blog nicht behandelt wird: Wie Sie zwei Streams nur mit SQL in einem Stream-Stream-Join zusammenführen!

Klonen Sie das Repository in Ihre Databricks-Umgebung, um loszulegen, oder verbinden Sie dbt Cloud kostenlos mit Databricks über Partner Connect. Sie können auch mehr mit der Dokumentation für Materialized Views und Streaming Tables erfahren.

(Dieser Blogbeitrag wurde mit KI-gestützten Tools übersetzt.) Originalbeitrag