Herstellen einer Verbindung mit Lakebase

Verwenden Sie Structured Streaming, um in Lakebase oder eine externe PostgreSQL-Datenbank mit integriertem Batching, automatischen Wiederholungen und arbeitsbereichsverwalteter Authentifizierung zu schreiben.

Wann der Lakebase-Sink verwendet werden sollte

Verwenden Sie den Lakebase-Sink für Streaming-Schreibvorgänge mit niedriger Latenz in Lakebase oder eine externe PostgreSQL-Datenbank. Für diesen Sink müssen Sie keine benutzerdefinierten foreach Funktionen für die Batchverarbeitung, Verbindungsverwaltung und Fehlerbehandlung implementieren.

Zu den gängigen Anwendungsfällen gehören:

• Aktualisieren Sie Anwendungsdatenbanken in Echtzeit für Betriebsdashboards oder kundenorientierte Features.
• Synchronisieren Sie fortlaufend geänderte Daten, z. B. aggregierte oder gefilterte Streamingergebnisse, in einer Transaktionsdatenbank.
• Schreiben Sie die Ausgabe einer Strukturierten Streaming-Abfrage in eine Lakebase-Tabelle mit Unter-Sekunde-Latenz im Echtzeitmodus.

Informationen zum Synchronisieren von Daten aus Lakebase zu Delta Lake-Tabellen im Lakehouse finden Sie in umgekehrter Richtung unter Lakebase Change Data Feed.

Anforderungen

• Databricks Runtime 18 LTS und höher.
  • Externe PostgreSQL-Verbindungen erfordern, dass Sie Databricks Runtime 19 und höher verwenden und sich für die benutzerdefinierte JDBC-Vorschau in UC Compute anmelden.
  • Intervalldatentypen erfordern, dass Sie Databricks Runtime 19 und höher verwenden.
• Klassische Rechenleistung mit dedizierten oder Standardzugriffsmodi oder serverlose Berechnung für Notebooks oder Jobs. Verwenden Sie für serverlose Rechenumgebungen Trigger.AvailableNow(). Siehe Streaming mit serverloser Rechenleistung.
• Eine Lakebase-Datenbank oder eine Unity-Catalog-Verbindung zu einer externen PostgreSQL-Datenbank.

Identifikationsanforderungen

Für alle Ziele empfiehlt Databricks die Verwendung von Schema-, Tabellen-, Spalten- und Primärschlüssel-Spaltennamen, die mit einem Buchstaben oder Unterstrich beginnen und nur Buchstaben, Zahlen und Unterstriche enthalten. Der Sink erzwingt diese Anforderungen, wenn er automatisch eine Lakebase-Tabelle erstellt. Um Kennungen zu verwenden, die diese Anforderungen nicht erfüllen, erstellen Sie vor Beginn der Abfrage die Zieltabelle.

Mit Datenbank verbinden

Die Lakebase-Spüle unterstützt die folgenden Verbindungsmethoden:

Lakebase-Tabellen, die im Unity-Katalog registriert sind

Bei Lakebase-Tabellen, die mit Unity Catalog registriert sind, verwaltet der Connector automatisch die Anmeldeinformationen und verwendet die Identität des Benutzers oder Dienstprinzipals, der die Abfrage ausführt. Wenn die Tabelle nicht vorhanden ist, erstellt der Verbinder die Tabelle.

Informationen zum Registrieren einer Lakebase-Datenbank im Unity-Katalog finden Sie unter Registrieren einer Lakebase-Datenbank im Unity-Katalog.

Um in eine Lakebase-Tabelle zu schreiben, verwendet man die Methode .toTable() mit einem vollständig qualifizierten Tabellennamen: catalog.schema.table

Python

(df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-columns>")  # Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")
)

Scala

df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-columns>")  // Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")

Ersetzen Sie die folgenden Platzhalter:

• <catalog>.<schema>.<table>: Der vollqualifizierte Name der Zieltabelle. Der catalog Unity-Katalogkatalog, den Sie beim Registrieren der Lakebase-Datenbank erstellt haben, finden Sie unter Registrieren einer Lakebase-Datenbank im Unity-Katalog. Wenn die Tabelle nicht vorhanden ist, erstellt der Verbinder sie.
• <primary-key-columns>:Wahlfrei. Eine kommagetrennte Liste aller Spalten im Primärschlüssel der Zieltabelle, zum Beispiel id oder user_id,event_type. Siehe Upsert-Verhalten.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Ein Unity Catalog-Volumepfad, in dem die Abfrage ihren Prüfpunkt speichert. Sie können auch einen Cloudobjektspeicher-URI verwenden. Der Speicherort muss Speicher sein, in den Sie schreiben können, nicht auf einen lokalen Datenträger, und muss für jede Streamingabfrage eindeutig sein. Dies ist unabhängig von der Zieltabelle. Siehe Prüfpunkte für strukturiertes Streaming.

Für optionale Konfigurationen wie batchsize und batchintervalsiehe die PostgreSQL-Sink-Optionen.

Lakebase-Tabellen, die nicht im Unity-Katalog registriert sind

Für Lakebase-Tabellen, die nicht im Unity-Katalog registriert sind, verwaltet der Connector automatisch die Anmeldeinformationen und verwendet die Identität des Benutzers oder Dienstprinzipals, der die Abfrage ausführt. Wenn die Tabelle nicht vorhanden ist, erstellt der Verbinder die Tabelle.

Verwenden Sie zum Schreiben in eine Lakebase-Tabelle die Optionen endpoint und dbtable:

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  # Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-columns>")  # Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()
)

Scala

df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  // Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-columns>")  // Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

Ersetzen Sie die folgenden Platzhalter:

• <project-id>.<branch-id>.<endpoint-id>: Ihr Lakebase-Endpunkt. Suchen Sie alle drei Werte im Ressourcennamen im Menü " Get ID " der Registerkarte " Computes ", die das Format projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>aufweist. Siehe Berechnen von Bezeichnern.
• <database>:Wahlfrei. Der Name der Ziel-PostgreSQL-Datenbank. Wird standardmäßig auf databricks_postgres festgelegt. Siehe "Datenbanken verwalten".
• <schema>.<table>: Die Zieltabelle im schema.table Format. Wenn Sie das Schema weglassen, verwendet die Spüle das public Schema. Für die automatische Tabellenerstellung verwenden Sie Identifikatoren, die mit einem Buchstaben oder Unterstrich beginnen und nur Buchstaben, Zahlen und Unterstriche enthalten.
• <primary-key-columns>:Wahlfrei. Eine kommagetrennte Liste aller Spalten im Primärschlüssel der Zieltabelle, zum Beispiel id oder user_id,event_type. Siehe Upsert-Verhalten.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Ein Unity Catalog-Volumepfad, in dem die Abfrage ihren Prüfpunkt speichert. Sie können auch einen Cloudobjektspeicher-URI verwenden. Der Speicherort muss Speicher sein, in den Sie schreiben können, nicht auf einen lokalen Datenträger, und muss für jede Streamingabfrage eindeutig sein. Dies ist unabhängig von der Zieltabelle. Siehe Prüfpunkte für strukturiertes Streaming.

Für optionale Konfigurationen wie batchsize und batchintervalsiehe die PostgreSQL-Sink-Optionen.

Externe PostgreSQL mit Unity-Katalog-Zugangsdaten

Verwenden Sie eine Unity-Catalog-Verbindung, um sich bei einer externen PostgreSQL-Datenbank zu authentifizieren, ohne Zugangsdaten in Ihrem Code zu speichern. Die Zieltabelle muss bereits existieren.

Stelle eine Verbindung vom Typ POSTGRESQLan, siehe Create a Connection. Der Benutzer oder Serviceprinzipal, der die Abfrage ausführt, muss für die Verbindung über USE CONNECTION verfügen.

Um in die PostgreSQL-Tabelle zu schreiben, verwenden Sie die databricks.connection, database, und dbtable Optionen:

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("databricks.connection", "<connection-name>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-columns>")  # Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()
)

Scala

df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("databricks.connection", "<connection-name>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-columns>")  // Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

Ersetzen Sie die folgenden Platzhalter:

• <connection-name>: Der Name der Unity Catalog-Verbindung.
• <database>: Der Name der Ziel-PostgreSQL-Datenbank.
• <schema>.<table>: Die bestehende Zieltabelle im Format schema.table . Wenn Sie das Schema weglassen, verwendet die Spüle das public Schema.
• <primary-key-columns>:Wahlfrei. Eine kommagetrennte Liste aller Spalten im Primärschlüssel der Zieltabelle, zum Beispiel id oder user_id,event_type. Siehe Upsert-Verhalten.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Ein Unity Catalog-Volumepfad, in dem die Abfrage ihren Prüfpunkt speichert. Sie können auch einen Cloudobjektspeicher-URI verwenden. Der Speicherort muss Speicher sein, in den Sie schreiben können, nicht auf einen lokalen Datenträger, und muss für jede Streamingabfrage eindeutig sein. Dies ist unabhängig von der Zieltabelle. Siehe Prüfpunkte für strukturiertes Streaming.

PostgreSQL-Verbindungen verwenden immer TLS. Die Zertifikatsverifizierung folgt den Einstellungen der Unity-Katalog-Verbindung, die Sie beim Erstellen der Verbindung auswählen:

• Vertrauensserverzertifikat: Wenn ausgewählt, verwendet sslmode=requiredie Verbindung , das die Verbindung verschlüsselt, ohne das Serverzertifikat zu überprüfen.
• Vom Nutzer bereitgestelltes Serverzertifikat: Stellen Sie ein PEM-codiertes Serverzertifikat bereit, das verwendet sslmode=verify-full wird, wenn das Trust-Server-Zertifikat nicht ausgewählt ist. Wenn du kein Zertifikat bereitstellst, verwendet die Verbindung sslmode=verify-full den JVM-Standard-Vertrauensspeicher.

Konfigurationsoptionen

Die Senke löst einen Fehler für nicht erkannte Optionen aus, JDBC_STREAMING_SINK_INVALID_OPTIONS.

Für die Sink-Konfigurationsoptionen, einschließlich der gemeinsamen Optionen und der Optionen für jede Verbindungsmethode, siehe die PostgreSQL-Sink-Optionen.

Datentypzuordnungen

Der Sink prüft, dass jede DataFrame-Spalte mit der entsprechenden Zielspalte kompatibel ist, bevor er in eine bestehende Lakebase- oder externe PostgreSQL-Tabelle schreibt.

Die folgende Tabelle enthält Typen, die in Databricks Runtime 18 LTS und höher unterstützt werden:

Die folgende Tabelle enthält Typen, die in Databricks Runtime 19 und höher unterstützt werden:

Upsert-Verhalten

Die upsertkey Option identifiziert die Primärschlüsselspalten der Zieltabelle. Für eine bestehende Tabelle müssen die Spalten in upsertkey exakt mit dem Primärschlüssel der Tabelle übereinstimmen. Wenn du die Option weglässt, liest das Waschbecken den Primärschlüssel aus der Tabelle. Für eine Lakebase-Tabelle, die vom Sink erstellt wird, definiert upsertkey den Primärschlüssel. Wenn du die Option weglässt, erstellt der Sink die Tabelle ohne einen Primärschlüssel.

Wenn die Zieltabelle einen Primärschlüssel hat, führt der Sink Upserts mithilfe der PostgreSQL-INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ...-Syntax aus. Wenn die Zieltabelle keinen Primärschlüssel hat, führt der Sink Einfügungen durch. Der Ausgabemodus einer Abfrage hat keinen Einfluss auf dieses Verhalten.

Alle Primärschlüsselspalten müssen im DataFrame vorhanden sein und vergleichbare Typen verwenden, wie numerische oder String-Typen.

Leistungsoptimierung

Batchverarbeitung und Rückdruck

Ein Flush wird ausgelöst, sobald eine der beiden Bedingungen erfüllt ist:

• Der Puffer erreicht batchsize Zeilen, was standardmäßig 1000 ist.
• Das Alter des Puffers überschreitet batchinterval, was standardmäßig auf 100 milliseconds festgelegt ist.

Wenn die Datenbank nicht mit der eingehenden Datenrate Schritt halten kann, propagiert der Sink Backpressure stromaufwärts an die Quelle.

Hinweise zu Latenz und Durchsatz:

• Verringern Sie für Workloads mit geringer Latenz im Echtzeitmodus batchinterval, um eine kürzere maximale Zeit vor dem Leeren zu gewährleisten. Siehe Echtzeit-Modus-Konzepte für Konzepte und Echtzeit-Modus-Beispiele für ein Codebeispiel.
• Erhöhen Sie batchsize für Workloads mit hohem Durchsatz, um den Overhead für jede Transaktion zu reduzieren.

Verbindungsverhalten

Die Senke verwendet Verbindungspooling auf Executors. Standardmäßig verwendet jede Aufgabe eine Datenbankverbindung.

Databricks empfiehlt, den Standardwert der 1 Aufgabe für jede Verbindung zu verwenden. Wenn Sie die Anzahl der Aufgaben pro Verbindung erhöhen, kann es zu Konkurrenz um Verbindungen kommen und die Latenzen bei Verbindungen mit hohem Durchsatz können steigen.

Um das Verhältnis von Vorgängen zu Verbindungen zu konfigurieren, legen Sie die spark.databricks.sql.streaming.jdbc.tasksPerConnection Spark-Konfiguration fest. Wenn die Zieldatenbank ein niedriges Limit für Verbindungen hat, verringern Sie die Anzahl der Shuffle-Partitionen oder erhöhen Sie spark.databricks.sql.streaming.jdbc.tasksPerConnection.

Die Senke wiederholt vorübergehende JDBC-Fehler automatisch, einschließlich Verbindungsfehlern, Deadlocks und Ratenbegrenzungen. Wenn der Sink alle Wiederholungsversuche ausgeschöpft hat, schlägt die Abfrage fehl.

Unterstützte Trigger und Ausgabemodi

Triggers

Diese Tabelle zeigt die Unterstützung für Structured Streaming-Triggertypen auf klassischer und serverloser Rechenleistung:

Ausgabemodi

Diese Tabelle zeigt die Unterstützung für die Ausgabemodi für strukturiertes Streaming:

Einschränkungen

• Für eine externe PostgreSQL-Datenbank, die über eine Unity-Catalog-Verbindung verbunden ist, muss die Zieltabelle bereits existieren. Der Sink erstellt fehlende Tabellen automatisch nur in Lakebase.
• Lakeflow-Pipelines werden nicht unterstützt.