Connettersi a Lakebase

Usare Structured Streaming per scrivere in Lakebase con batch predefiniti, tentativi automatici e autenticazione gestita dall'area di lavoro.

Quando usare il sink Lakebase

Usare il sink Lakebase per le scritture di streaming a bassa latenza in Lakebase. Questo sink non richiede l'implementazione di funzioni personalizzate foreachBatch per gestire l'invio in batch, la gestione delle connessioni e la gestione degli errori.

I casi d'uso comuni includono:

• Aggiornare i database dell'applicazione in tempo reale per dashboard operativi o funzionalità rivolte ai clienti.
• Sincronizzare in modo continuo i dati, ad esempio i risultati di streaming aggregati o filtrati, in un database transazionale.
• Scrivi l'output di una query di Structured Streaming in una tabella Lakebase con latenza inferiore al secondo usando la modalità in tempo reale.

Per sincronizzare i dati da Lakebase verso le tabelle Delta Lake nel Lakehouse, nella direzione inversa, vedere Lakebase Change Data Feed.

Requisiti

• Databricks Runtime 18 e versioni successive
• Calcolo classico con modalità di accesso dedicate o standard.
• Un database Lakebase

Connettersi a un database

Il sink Lakebase supporta i metodi di connessione seguenti:

Tabelle Lakebase registrate con Unity Catalog

Per le tabelle Lakebase registrate con Unity Catalog, il connettore gestisce automaticamente le credenziali e utilizza l'identità dell'utente o del service principal che esegue la query. Se la tabella non esiste, il connettore crea la tabella.

Per registrare un database Lakebase con Il catalogo Unity, vedere Registrare un database Lakebase nel catalogo unity.

Per scrivere in una tabella Lakebase, utilizzare il metodo .toTable() con un nome tabella completamente qualificato, catalog.schema.table. L'esempio seguente mostra le opzioni necessarie, oltre all'opzione facoltativa upsertkey :

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>")

Sostituire i segnaposto seguenti:

• <catalog>.<schema>.<table>: Il nome completo della tabella di destinazione. catalog è il catalogo di Unity Catalog che hai creato quando hai registrato il database Lakebase; vedi Registrare un database Lakebase in Unity Catalog. Se la tabella non esiste, il connettore lo crea.
• <primary-key-column>: facoltativo. Elenco delimitato da virgole delle colonne che formano la chiave upsert, ad esempio id o user_id,event_type. Se si omette upsertkey, il sink deduce la chiave dalla chiave primaria della tabella di destinazione. Vedere Comportamento di Upsert.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: percorso di un volume di Unity Catalog in cui la query archivia il checkpoint. È anche possibile usare un URI di archiviazione di oggetti cloud. La posizione deve essere un'unità di archiviazione su cui sia possibile scrivere, non un disco locale, e deve essere univoca per ogni query di streaming. Questo è indipendente dalla tabella di destinazione. Consulta Checkpoint di Structured Streaming.

Per le configurazioni facoltative, ad esempio batchsize e batchinterval, vedere Opzioni di configurazione.

Tabelle Lakebase non registrate con Unity Catalog

Per le tabelle Lakebase non registrate in Unity Catalog, il connettore gestisce automaticamente le credenziali e usa l'identità dell'utente o dell'entità servizio che esegue la query. Se la tabella non esiste, il connettore crea la tabella.

Per scrivere in una tabella di Lakebase, usa le opzioni endpoint e dbtable. L'esempio seguente include anche le opzioni facoltative database e upsertkey:

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()

Sostituire i segnaposto seguenti:

• <project-id>.<branch-id>.<endpoint-id>: Il tuo endpoint Lakebase. Trovare tutti e tre i valori nel nome della risorsa nel menu Recupera ID della scheda Calcolo , che ha il formato projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>. Vedere Identificatori di calcolo.
• <database>: facoltativo. Nome del database Postgres di destinazione. Di default è databricks_postgres. Vedere Gestire i database.
• <schema>.<table>: La tabella di destinazione in formato schema.table. Se si omette lo schema, il sink usa lo public schema . Usa identificatori semplici che iniziano con una lettera o un carattere di sottolineatura e contengono solo lettere, numeri e caratteri di sottolineatura; Gli identificatori delimitati e i caratteri speciali, ad esempio trattini, non sono supportati.
• <primary-key-column>: facoltativo. Elenco delimitato da virgole delle colonne che formano la chiave upsert, ad esempio id o user_id,event_type. Se si omette upsertkey, il sink deduce la chiave dalla chiave primaria della tabella di destinazione. Vedere Comportamento di Upsert.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: percorso di un volume di Unity Catalog in cui la query archivia il checkpoint. È anche possibile usare un URI di archiviazione di oggetti cloud. La posizione deve essere un'unità di archiviazione su cui sia possibile scrivere, non un disco locale, e deve essere univoca per ogni query di streaming. Questo è indipendente dalla tabella di destinazione. Consulta Checkpoint di Structured Streaming.

Per le configurazioni facoltative, ad esempio batchsize e batchinterval, vedere Opzioni di configurazione.

Opzioni di configurazione

Il sink segnala un errore in presenza di opzioni non riconosciute, JDBC_STREAMING_SINK_INVALID_OPTIONS.

Le opzioni seguenti si applicano a tutti i metodi di connessione:

Tabelle Lakebase non registrate con Unity Catalog

Quando ci si connette a una tabella Lakebase non registrata in Unity Catalog, si applicano le opzioni seguenti:

Comportamento di upsert

Quando esistono chiavi di upsert, specificate con upsertkey o dedotte dal sink dalle chiavi primarie della tabella, il sink esegue un upsert nella tabella utilizzando la sintassi INSERT INTO ... ON CONFLICT (<upsert_key>) DO UPDATE SET ... di PostgreSQL.

Quando non esistono chiavi upsert, il sink esegue inserimenti. La modalità di output di una query non ha alcun effetto sul comportamento dell'upsert o dell'inserimento.

Le upsertkey colonne devono:

• Essere un subset non vuoto delle colonne del dataframe.
• Fai corrispondere esattamente la PRIMARY KEY della tabella di destinazione. Se le colonne specificate non corrispondono alla chiave primaria, la query ha esito negativo.
• Essere tipi confrontabili, ad esempio tipi numerici o stringa. Per evitare deadlock nel database durante le scritture concorrenti, il sink ordina le righe in base alla chiave di upsert all'interno di ciascun batch. Le chiavi Upsert non supportano tipi complessi o struct.

I nomi delle colonne vengono automaticamente racchiusi tra virgolette con il valore predefinito di PostgreSQL, le virgolette doppie ", per gestire le parole chiave riservate e i nomi con maiuscole e minuscole.

I nomi di tabella e schema devono usare identificatori semplici che iniziano con una lettera o un carattere di sottolineatura e contengono solo lettere, numeri e caratteri di sottolineatura. Il sink non supporta identificatori tra virgolette né caratteri speciali, come i trattini, nei nomi di tabelle o schemi.

Ottimizzazione delle prestazioni

Invio in batch e backpressione

Lo svuotamento viene eseguito quando è soddisfatta una delle due condizioni:

• Il buffer arriva a batchsize righe, il cui valore predefinito è 1000.
• L'età del buffer supera batchinterval, che per impostazione predefinita è 100 milliseconds.

Quando il database non riesce a tenere il passo con il tasso di dati in ingresso, il sink propaga la backpressure a monte, fino alla sorgente.

Indicazioni sulla latenza e sulla velocità effettiva:

• Per i carichi di lavoro a bassa latenza in modalità in tempo reale, diminuire batchinterval per garantire un intervallo massimo più breve prima dello svuotamento. Vedere Modalità in tempo reale in Structured Streaming per concetti ed esempi di modalità in tempo reale per un esempio di codice.
• Per i carichi di lavoro a throughput elevato, aumentare batchsize per ridurre il sovraccarico di ogni transazione.

Comportamento della connessione

Il "sink" utilizza un pool di connessioni sugli executor. Per impostazione predefinita, ogni attività usa una connessione al database.

Databricks consiglia di usare il valore predefinito dell'attività 1 per ogni connessione. Se si aumenta il numero di attività per ogni connessione, è possibile che si verifichino conflitti di connessione e si aumentino le latenze per le connessioni a velocità effettiva elevata.

Per configurare il rapporto tra attività e connessioni, impostare la spark.databricks.sql.streaming.jdbc.tasksPerConnection configurazione di Spark. Se il database di destinazione ha un limite di connessioni basso, ridurre il numero di partizioni di shuffle o aumentare spark.databricks.sql.streaming.jdbc.tasksPerConnection.

Il sink ritenta automaticamente gli errori JDBC temporanei, inclusi errori di connessione, deadlock e limitazione della frequenza. Se il sink esaurisce tutti i nuovi tentativi, la query ha esito negativo.

Trigger e modalità di output supportati

Triggers

Questa tabella mostra il supporto per i tipi di trigger Structured Streaming:

Modalità di output

Questa tabella mostra il supporto per le modalità di output di Structured Streaming:

Limitazioni

• Le pipeline di calcolo serverless e Lakeflow non sono supportate.
• Solo Lakebase è supportato come destinazione di scrittura. I database esterni compatibili con PostgreSQL non sono supportati.