Herstellen einer Verbindung mit Lakebase

Important

Dieses Feature befindet sich in der Public Preview.

Verwenden Sie strukturiertes Streaming zum Schreiben in Lakebase mit integrierter Batchverarbeitung, automatischen Wiederholungen und vom Arbeitsbereich verwalteter Authentifizierung.

Wann der Lakebase-Sink verwendet werden sollte

Verwenden Sie den Lakebase-Sink für Streaming-Schreibvorgänge mit geringer Latenz nach Lakebase. Für diesen Sink müssen Sie keine benutzerdefinierten foreachBatch 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 und höher
• Klassische Berechnung mit dedizierten oder Standardzugriffsmodi.
• Eine Lakebase-Datenbank

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.

Verwenden Sie die .toTable() Methode mit einem vollqualifizierten Tabellennamen, um in eine Lakebase-Tabelle zu schreiben. catalog.schema.table Das folgende Beispiel zeigt die erforderlichen Optionen sowie die optionale upsertkey Option:

Python

(df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")
)

Scala

df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .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-column>:Wahlfrei. Eine durch Trennzeichen getrennte Liste der Spalten, die den Upsert-Schlüssel bilden, z. B id . oder user_id,event_type. Wenn Sie upsertkey weglassen, leitet die Senke den Schlüssel aus dem Primärschlüssel der Zieltabelle ab. 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.

Informationen zu optionalen Konfigurationen wie batchsize und batchinterval finden Sie unter Konfigurationsoptionen.

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.

Um in eine Lakebase-Tabelle zu schreiben, verwenden Sie die Optionen endpoint und dbtable. Das folgende Beispiel enthält auch die optionalen database und upsertkey optionen:

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-column>")  # Optional. Inferred from the table's primary key if omitted.
  .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-column>")  // Optional. Inferred from the table's primary key if omitted.
  .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 Zieldatenbank Postgres. 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. Verwenden Sie einfache Bezeichner, die mit einem Buchstaben oder Unterstrich beginnen und nur Buchstaben, Zahlen und Unterstriche enthalten; in Anführungszeichen gesetzte Bezeichner und Sonderzeichen, z. B. Bindestriche, werden nicht unterstützt.
• <primary-key-column>:Wahlfrei. Eine durch Trennzeichen getrennte Liste der Spalten, die den Upsert-Schlüssel bilden, z. B id . oder user_id,event_type. Wenn Sie upsertkey weglassen, leitet die Senke den Schlüssel aus dem Primärschlüssel der Zieltabelle ab. 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.

Informationen zu optionalen Konfigurationen wie batchsize und batchinterval finden Sie unter Konfigurationsoptionen.

Konfigurationsoptionen

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

Die folgenden Optionen gelten für alle Verbindungsmethoden:

Schlüssel	Standard	Beschreibung
batchinterval	100 milliseconds	Wahlfrei. Die maximale Zeit, die Zeilen im Puffer gehalten werden, bevor sie geleert werden. Beispiel: "50 milliseconds"
batchsize	1000	Wahlfrei. Die maximale Anzahl von Zeilen für jede Datenbanktransaktion.
checkpointLocation	Nichts	Erforderlich Pfad zu einem Prüfpunktverzeichnis, z. B. einem Unity-Katalogvolume (/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>). Muss für jede Abfrage eindeutig sein. Siehe Prüfpunkte für strukturiertes Streaming.
upsertkey	Nichts	Wahlfrei. Eine durch Kommas getrennte Liste von Spaltennamen, die den Upsert-Schlüssel bilden. Zum Beispiel: "id" oder "user_id,event_type". Wenn Sie angeben upsertkey, müssen die Spalten mit dem Primärschlüssel der Tabelle übereinstimmen, oder die Abfrage schlägt fehl. Wenn Sie es weglassen, verwendet die Spüle automatisch den Primärschlüssel. Weitere Informationen finden Sie unter Upsert-Verhalten.

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

Die folgenden Optionen gelten, wenn Sie eine Verbindung mit einer Lakebase-Tabelle herstellen, die nicht im Unity-Katalog registriert ist:

Schlüssel	Standard	Beschreibung
database	databricks_postgres	Wahlfrei. Der Zielname der PostgreSQL-Datenbank.
dbtable	Nichts	Erforderlich Der Name der Zieltabelle im schema.table Format. Wenn Sie kein Schema angeben, lautet publicder Standardwert . Verwenden Sie einfache Bezeichner, die mit einem Buchstaben oder Unterstrich beginnen und nur Buchstaben, Zahlen und Unterstriche enthalten. Setzen Sie Tabellen- oder Schemanamen nicht in Anführungszeichen. In Anführungszeichen gesetzte Bezeichner und Namen mit Sonderzeichen, z. B. Bindestrichen, werden nicht unterstützt.
endpoint	Nichts	Erforderlich Der Lakebase-Endpunkt im Format project_id.branch_id oder project_id.branch_id.endpoint_id. Der endpoint_id ist optional. Wenn Sie ihn weglassen und die Verzweigung über einen einzelnen Lese-/Schreibendpunkt verfügt, wählt die Senke diesen Endpunkt standardmäßig aus.

Upsert-Verhalten

Wenn Upsert-Schlüssel vorhanden sind, die entweder mit upsertkey angegeben oder von der Senke aus den Primärschlüsseln der Tabelle abgeleitet wurden, führt die Senke einen Upsert in die Tabelle mit der INSERT INTO ... ON CONFLICT (<upsert_key>) DO UPDATE SET ...-Syntax von PostgreSQL durch.

Wenn keine Upsert-Schlüssel vorhanden sind, führt die Senke Einfügungen durch. Der Ausgabemodus einer Abfrage hat keine Auswirkungen auf das Upsert- oder Einfügeverhalten.

Die upsertkey Spalten müssen:

• Eine nicht leere Teilmenge der DataFrame-Spalten sein.
• Passen Sie PRIMARY KEY genau an die Zieltabelle an. Wenn die von Ihnen angegebenen Spalten nicht mit dem Primärschlüssel übereinstimmen, schlägt die Abfrage fehl.
• Seien Sie vergleichbare Typen, z. B. numerische oder Zeichenfolgentypen. Um Datenbankdeadlocks bei gleichzeitigen Schreibvorgängen zu verhindern, sortiert die Senke Zeilen nach Upsert-Schlüssel innerhalb jedes Batches. Upsert-Schlüssel unterstützen keine komplexen oder Strukturtypen.

Spaltennamen werden automatisch mit dem PostgreSQL-Standard – doppelten Anführungszeichen " – in Anführungszeichen gesetzt, wodurch reservierte Schlüsselwörter und Namen mit gemischter Groß- und Kleinschreibung berücksichtigt werden.

Tabellen- und Schemanamen müssen einfache Bezeichner verwenden, die mit einem Buchstaben oder Unterstrich beginnen und nur Buchstaben, Zahlen und Unterstriche enthalten. Die Senke unterstützt keine in Anführungszeichen gesetzten Bezeichner oder Sonderzeichen, z. B. Bindestriche, in Tabellen- oder Schemanamen.

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. Informationen zu den Konzepten finden Sie unter Echtzeitmodus in Structured Streaming; ein Codebeispiel finden Sie unter Beispiele zum Echtzeitmodus.
• 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 strukturierte Streamingtriggertypen:

Auslöser	Unterstützt
realTime	Yes
ProcessingTime	Yes
AvailableNow	Yes
Once	Yes

Ausgabemodi

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

Ausgabemodus	Unterstützt
update	Yes
append	Yes. Das Verhalten ist identisch mit update. Die Abfrage führt einen Upsert durch, wenn die Zieltabelle über einen Primärschlüssel verfügt, andernfalls führt die Abfrage eine Einfügung durch. Siehe Upsert-Verhalten.
complete	No

Einschränkungen

• Serverlose Compute- und Lakeflow-Pipelines werden nicht unterstützt.
• Nur Lakebase wird als Schreibziel unterstützt. Externe PostgreSQL-kompatible Datenbanken werden nicht unterstützt.