Einführung der DLT Sink API: Pipelines nach Kafka und externe Delta-Tabellen schreiben

Wenn Sie neu bei Delta Live Tables sind, empfehlen wir Ihnen, vor dem Lesen dieses Blogs Getting Started with Delta Live Tables zu lesen. Dort wird erklärt, wie Sie skalierbare und zuverlässige Pipelines mit den deklarativen ETL-Definitionen und -Anweisungen von Delta Live Tables (DLT) erstellen können.

Einleitung

Delta Live Tables (DLT) Pipelines bieten eine robuste Plattform für die Erstellung zuverlässiger, wartbarer und testbarer Datenverarbeitungs-Pipelines innerhalb von Databricks. Durch die Nutzung seines deklarativen Frameworks und die automatische Bereitstellung optimaler Serverless-Compute vereinfacht DLT die Komplexität von Streaming, Datentransformation und -verwaltung und liefert Skalierbarkeit und Effizienz für moderne Daten-Workflows.

Traditionell boten DLT Pipelines eine effiziente Möglichkeit, Daten entweder als Streaming Tables oder Materialized Views, die von Unity Catalog verwaltet werden, zu erfassen und zu verarbeiten. Während dieser Ansatz die meisten Datenverarbeitungsanforderungen erfüllt, gibt es Fälle, in denen Datenpipelines mit externen Systemen verbunden werden müssen oder Structured Streaming Sinks verwendet werden müssen, anstatt in Streaming Tables oder Materialized Views zu schreiben.

Die Einführung der neuen Sinks API in DLT adressiert dies, indem sie es Benutzern ermöglicht, verarbeitete Daten in externe Event-Streams wie Apache Kafka, Azure Event Hubs zu schreiben, sowie in eine Delta Table zu schreiben. Diese neue Funktion erweitert den Umfang von DLT Pipelines und ermöglicht eine nahtlose Integration mit externen Plattformen.

Diese Funktionen befinden sich jetzt in der Public Preview und wir werden im Laufe der Zeit weitere Sinks von Databricks Runtime zu DLT hinzufügen und schließlich alle unterstützen. Der nächste Punkt, an dem wir arbeiten, ist foreachBatch, das es Kunden ermöglicht, in beliebige Datensenken zu schreiben und benutzerdefinierte Merges in Delta-Tabellen durchzuführen.

Die Sink API ist im Python-Paket dlt verfügbar und kann wie unten gezeigt mit create_sink() verwendet werden:

Die API akzeptiert drei Schlüsselargumente zur Definition des Sinks:

• Sink-Name: Eine Zeichenkette, die den Sink eindeutig innerhalb Ihrer Pipeline identifiziert. Dieser Name ermöglicht es Ihnen, den Sink zu referenzieren und zu verwalten.
• Format-Spezifikation: Eine Zeichenkette, die das Ausgabeformat bestimmt, mit Unterstützung für entweder "kafka" oder "delta".
• Sink-Optionen: Ein Dictionary von Schlüssel-Wert-Paaren, bei denen sowohl Schlüssel als auch Werte Zeichenketten sind. Für Kafka-Sinks können alle in Structured Streaming verfügbaren Konfigurationsoptionen genutzt werden, einschließlich Einstellungen für Authentifizierung, Partitionierungsstrategien und mehr. Eine umfassende Liste der von Kafka unterstützten Konfigurationsoptionen finden Sie in den Dokumenten. Delta-Sinks bieten eine einfachere Konfiguration, indem sie es Ihnen ermöglichen, entweder einen Speicherpfad mit dem Attribut path zu definieren oder direkt in eine Tabelle in Unity Catalog mit dem Attribut tableName zu schreiben.

Schreiben in einen Sink

Die @append_flow API wurde erweitert, um das Schreiben von Daten in Ziel-Sinks zu ermöglichen, die durch ihre Sink-Namen identifiziert werden. Traditionell ermöglichte diese API Benutzern, nahtlos Daten aus mehreren Quellen in eine einzige Streaming-Tabelle zu laden. Mit der neuen Erweiterung können Benutzer jetzt auch Daten an bestimmte Sinks anhängen. Nachfolgend finden Sie ein Beispiel, das zeigt, wie dies eingerichtet wird:

Erstellung der Pipeline

Lassen Sie uns nun eine DLT-Pipeline erstellen, die Clickstream-Daten verarbeitet, die in den Databricks-Datasets verpackt sind. Diese Pipeline wird die Daten parsen, um Ereignisse zu identifizieren, die mit einer Apache Spark-Seite verknüpft sind, und diese Daten anschließend sowohl an Event Hubs als auch an Delta-Sinks schreiben. Wir werden die Pipeline mithilfe der Medallion-Architektur strukturieren, die Daten in verschiedene Schichten organisiert, um Qualität und Verarbeitungseffizienz zu verbessern.

Wir beginnen damit, rohe JSON-Daten mithilfe von Auto Loader in die Bronze-Schicht zu laden. Anschließend bereinigen wir die Daten und erzwingen Qualitätsstandards in der Silver-Schicht, um deren Integrität sicherzustellen. Schließlich filtern wir in der Gold-Schicht Einträge mit einem aktuellen Seitentitel von Apache_Spark und speichern sie in einer Tabelle namens spark_referrers, die als Quelle für unsere Sinks dient. Das vollständige Codebeispiel finden Sie im Anhang.

Konfiguration des Azure Event Hubs Sinks

In diesem Abschnitt verwenden wir die create_sink API, um einen Event Hubs Sink einzurichten. Dies setzt voraus, dass Sie einen betriebsbereiten Kafka- oder Event Hubs-Stream haben. Unsere Pipeline wird Daten mithilfe einer Shared Access Policy in Kafka-fähige Event Hubs streamen, wobei die Verbindungszeichenfolge sicher in Databricks Secrets gespeichert ist. Alternativ können Sie für die Integration einen Service Principal anstelle einer SAS-Richtlinie verwenden. Stellen Sie sicher, dass Sie die Verbindungseigenschaften und Geheimnisse entsprechend aktualisieren. Hier ist der Code zur Konfiguration des Event Hubs Sinks:

Konfiguration des Delta Sinks

Zusätzlich zum Event Hubs Sink können wir die create_sink API verwenden, um einen Delta Sink einzurichten. Dieser Sink schreibt Daten an einen bestimmten Speicherort im Databricks File System (DBFS), kann aber auch so konfiguriert werden, dass er an einen Objektspeicherort wie Amazon S3 oder ADLS schreibt.

Nachfolgend finden Sie ein Beispiel für die Konfiguration eines Delta Sinks:

Erstellung von Flows zur Befüllung von Kafka- und Delta-Sinks

Nachdem die Event Hubs und Delta Sinks eingerichtet sind, besteht der nächste Schritt darin, diese Sinks mithilfe des append_flow Dekorators zu befüllen. Dieser Prozess beinhaltet das Streamen von Daten in die Sinks, um sicherzustellen, dass sie kontinuierlich mit den neuesten Informationen aktualisiert werden.

Für den Event Hubs Sink ist der Parameter value obligatorisch, während zusätzliche Parameter wie key, partition, headers und topic optional angegeben werden können. Nachfolgend finden Sie Beispiele für die Einrichtung von Flows für die Kafka- und Delta-Sinks:

Die Funktion applyInPandasWithState wird jetzt auch in DLT unterstützt, wodurch Benutzer die Leistung von Pandas für zustandsbehaftete Verarbeitung innerhalb ihrer DLT-Pipelines nutzen können. Diese Erweiterung ermöglicht komplexere Datentransformationen und Aggregationen mit der vertrauten Pandas API. Mit der DLT Sink API können Benutzer diese zustandsbehafteten verarbeiteten Daten einfach an Kafka-Topics streamen. Diese Integration ist besonders nützlich für Echtzeit-Analyse- und ereignisgesteuerte Architekturen, um sicherzustellen, dass Datenpipelines Streaming-Daten effizient an externe Systeme verarbeiten und verteilen können.

Alles zusammenbringen

Der oben gezeigte Ansatz demonstriert, wie eine DLT-Pipeline erstellt wird, die Daten effizient transformiert und gleichzeitig die neue Sink API nutzt, um die Ergebnisse nahtlos an externe Delta Tables und Kafka-fähige Event Hubs zu liefern.

Diese Funktion ist besonders wertvoll für Echtzeit-Analyse-Pipelines, da sie es ermöglicht, Daten in Kafka-Streams für Anwendungen wie Anomalieerkennung, vorausschauende Wartung und andere zeitkritische Anwendungsfälle zu streamen. Sie ermöglicht auch ereignisgesteuerte Architekturen, bei denen nachgelagerte Prozesse sofort durch das Streamen von Ereignissen in Kafka-Topics ausgelöst werden können, was eine schnelle Verarbeitung neu eingetroffener Daten ermöglicht.

Aufruf zur Handlung

Die DLT Sinks-Funktion ist jetzt für alle Databricks-Kunden in der Public Preview verfügbar! Diese leistungsstarke neue Funktion ermöglicht es Ihnen, Ihre DLT-Pipelines nahtlos auf externe Systeme wie Kafka und Delta-Tabellen zu erweitern und so Echtzeit-Datenflüsse und optimierte Integrationen sicherzustellen. Weitere Informationen finden Sie in den folgenden Ressourcen:

• Databricks Docs zu DLT Sinks
• Sehen Sie sich die Demo an zu Kafka Sinks in DLT mit Confluent Cloud
• Python Sprachreferenz für DLT

Anhang:

Pipeline-Code:

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