Connettersi a Lakebase

Usa Structured Streaming per scrivere su Lakebase o su un database esterno PostgreSQL con batching integrato, tentativi automatici e autenticazione gestita dall'area di lavoro.

Quando usare il sink Lakebase

Usa il sink di Lakebase per scritture in streaming a bassa latenza su Lakebase o su un database esterno PostgreSQL. Questo sink non richiede l'implementazione di funzioni personalizzate foreach 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 LTS e versioni successive.
  • Le connessioni PostgreSQL esterne richiedono l'utilizzo di Databricks Runtime 19 o versioni successive e l'adesione all'anteprima di Custom JDBC on UC Compute.
  • I tipi di dati di intervallo richiedono Databricks Runtime 19 o versione successiva.
  • Le strategie di merge per gli upsert richiedono di usare Databricks Runtime 19 e versioni successive.
• Calcolo classico con modalità di accesso dedicate o standard, oppure calcolo serverless per notebook o lavori. Sul calcolo serverless, si usa Trigger.AvailableNow(). Consulta Streaming sull'elaborazione serverless.
• Un database Lakebase, o una connessione Unity Catalog a un database esterno PostgreSQL.

Requisiti per l'identificazione

Per tutti i target, Databricks raccomanda di utilizzare nomi di schema, tabelle, colonne e colonne di chiave primaria che iniziano con una lettera o sottolineamento e contengono solo lettere, numeri e sottolineature. Il sink fa rispettare questi requisiti quando crea automaticamente una tabella Lakebase. Per utilizzare identificatori che non soddisfano questi requisiti, crea la tabella target prima di iniziare la query.

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 su una tabella Lakebase, usa il .toTable() metodo con un nome di tabella completamente qualificato, 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>")

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-columns>: facoltativo. Un elenco separato da virgole di tutte le colonne nella chiave primaria della tabella di destinazione, ad esempio id o user_id,event_type. Vedi 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 configurazioni opzionali, come batchsize e batchinterval, consulta le opzioni di sink di PostgreSQL.

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 su una tabella Lakebase, usa le opzioni endpoint e 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()

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. Il nome del database postgreSQL 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 . Per la creazione automatica di tabelle, usa identificatori che iniziano con una lettera o un sottolineamento e contengono solo lettere, numeri e sottolineature.
• <primary-key-columns>: facoltativo. Un elenco separato da virgole di tutte le colonne nella chiave primaria della tabella di destinazione, ad esempio id o user_id,event_type. Vedi 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 configurazioni opzionali, come batchsize e batchinterval, consulta le opzioni di sink di PostgreSQL.

PostgreSQL esterno con credenziali del Catalogo Unity

Usa una connessione Unity Catalog per autenticarti a un database PostgreSQL esterno senza memorizzare credenziali nel tuo codice. La tabella di destinazione deve già esistere.

Crea una connessione di tipo POSTGRESQL, vedi Crea una connessione. L'utente o il principale del servizio che esegue la query deve avere USE CONNECTION sulla connessione.

Per scrivere nella tabella PostgreSQL, usa le databricks.connectionopzioni , database, e dbtable :

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

Sostituire i segnaposto seguenti:

• <connection-name>: Il nome della connessione di Unity Catalog.
• <database>: Il nome del database postgreSQL di destinazione.
• <schema>.<table>: La tabella di destinazione esistente in formato schema.table. Se si omette lo schema, il sink usa lo public schema .
• <primary-key-columns>: facoltativo. Un elenco separato da virgole di tutte le colonne nella chiave primaria della tabella di destinazione, ad esempio id o user_id,event_type. Vedi 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.

Le connessioni PostgreSQL usano sempre TLS. La verifica del certificato segue le impostazioni sulla connessione Unity Catalog, che scegli quando crei la connessione:

• Certificato server fiduciario: Quando selezionato, la connessione utilizza sslmode=require, che cripta la connessione senza verificare il certificato server.
• Certificato server fornito dall'utente: Fornire un certificato server codificato in PEM da utilizzare sslmode=verify-full quando il certificato Trust Server non è selezionato. Se non fornisci un certificato, la connessione viene utilizzata sslmode=verify-full con il trust store predefinito della JVM.

Opzioni di configurazione

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

Per le opzioni di configurazione del sink, incluse le opzioni comuni e quelle per ogni metodo di connessione, vedi le opzioni del sink di PostgreSQL.

Mappatura dei tipi di dati

Il sink verifica che ogni colonna DataFrame sia compatibile con la colonna target corrispondente prima di scrivere su una Lakebase esistente o su una tabella esterna PostgreSQL.

La tabella seguente contiene i tipi supportati in Databricks Runtime 18 LTS e superiori:

La seguente tabella contiene i tipi supportati in Databricks Runtime 19 e superiori:

Comportamento da Upsert

L'opzione upsertkey identifica le colonne chiave principali della tabella di destinazione. Per una tabella esistente, le colonne in upsertkey devono corrispondere esattamente alla chiave primaria della tabella. Se ometti questa opzione, il sink legge la chiave primaria dalla tabella. Per una tabella Lakebase creata dal sink, upsertkey definisce la chiave primaria. Se ometti questa opzione, il sink crea la tabella senza chiave primaria.

Quando la tabella target ha una chiave primaria, il sink esegue un upsert utilizzando la sintassi INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ... di PostgreSQL. Quando la tabella di destinazione non ha una chiave primaria, il sink esegue inserimenti. La modalità di output di una query non ha alcun effetto su questo comportamento.

Tutte le colonne chiave primarie devono essere presenti nel DataFrame e utilizzare tipi comparabili, come tipi numerici o di stringhe.

Per controllare come gli upsert aggiornano le colonne non di chiave primaria, vedi Strategie di merge per gli upsert.

Strategie di fusione per gli upsert

In Databricks Runtime 19 e superiori, puoi usare strategie di merge per controllare come il sink aggiorna le colonne che non fanno parte della chiave primaria quando le righe hanno la stessa chiave primaria. Ad esempio, puoi mantenere il timestamp più grande o aggiornare una colonna solo quando un evento in arrivo ha un valore di ordine più recente.

Le strategie di fusione si applicano solo agli upsert. Se la tabella di destinazione non ha una chiave primaria, l'impostazione di upsert.mergeStrategies o upsert.defaultMergeStrategy causa un errore. Puoi assegnare strategie solo alle colonne non chiave primaria. Imposta upsert.mergeStrategies su un elenco separato da virgole nel formato column:strategy. Imposta upsert.defaultMergeStrategy per configurare un valore predefinito per le colonne senza strategie esplicite. upsert.defaultMergeStrategy non accetta latestBy o earliestBy. Per impostazione predefinita, il sink usa la strategia latest.

Strategie di fusione supportate

Come latestBy e earliestBy gestiscono i pareggi e i valori nulli

latestBy(column) e earliestBy(column) scelgono un valore confrontando il valore di ordine in ingresso con quello già memorizzato nella tabella. Per mantenere accurato quel valore memorizzato, la colonna di ordinazione deve utilizzare la strategia di abbinamento, o in upsert.mergeStrategies o attraverso upsert.defaultMergeStrategy:

• Per latestBy(column), la colonna di ordinamento deve usare greatest.
• Per earliestBy(column), la colonna di ordinamento deve usare least.

Ad esempio, is_success:latestBy(last_updated_at) richiede last_updated_at:greatest.

Quando entrambi i valori di ordinamento sono non nulli, latestBy(column) aggiorna una colonna solo quando il valore di ordinamento in ingresso è strettamente maggiore (>) del valore di ordinamento esistente. Per earliestBy(column), deve essere strettamente minore (<). Per entrambe le strategie, quando i valori di ordinamento sono uguali, il sink mantiene il valore esistente della colonna. All'interno di un singolo lotto, se più righe hanno la stessa chiave primaria e valore di ordine, il sink mantiene la prima riga che riceve.

Ad esempio, se is_success usa latestBy(last_updated_at) e una riga in ingresso ha lo stesso last_updated_at della riga esistente, il sink mantiene l'esistente is_success, anche se il valore in ingresso è diverso.

Per entrambe le strategie, i valori di ordinamento non nulli hanno la precedenza sui valori di ordinamento nullo. Una riga in ingresso con un valore di ordinamento non nullo sostituisce un valore esistente il cui valore di ordinamento è nullo. Una riga in ingresso con un valore di ordinamento nullo non sostituisce un valore esistente, anche quando il valore di ordinamento esistente è anch'esso nullo.

Esempio: modello logica upsert personalizzata

Si consideri una tabella con le colonne id, last_updated_at, creation_date, e is_success. Potresti voler esprimere la seguente logica:

INSERT INTO events ...
ON CONFLICT (id) DO UPDATE
SET
  -- Keep the most recent timestamp.
  last_updated_at = GREATEST(events.last_updated_at, EXCLUDED.last_updated_at),

  -- Keep the earliest non-null creation date.
  creation_date = COALESCE(events.creation_date, EXCLUDED.creation_date),

  -- Keep the value of is_success from the event with the most recent last_updated_at.
  is_success = CASE
    WHEN EXCLUDED.last_updated_at IS NOT NULL
      AND (events.last_updated_at IS NULL
        OR EXCLUDED.last_updated_at > events.last_updated_at)
    THEN EXCLUDED.is_success
    ELSE events.is_success
  END;

Per modellare questa logica, si utilizzano le seguenti strategie di fusione:

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("databricks.connection", "<connection-name>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "id")
  .option("upsert.mergeStrategies", "last_updated_at:greatest,creation_date:earliestNonNull,is_success:latestBy(last_updated_at)")
  .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", "id")
  .option("upsert.mergeStrategies", "last_updated_at:greatest,creation_date:earliestNonNull,is_success:latestBy(last_updated_at)")
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

Messa a punto 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. Vedi concetti di modalità in tempo reale per i concetti e 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 di Structured Streaming sia nel calcolo classico che serverless:

Modalità di output

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

Limitazioni

• Per un database PostgreSQL esterno collegato tramite una connessione Unity Catalog, la tabella target deve già esistere. Il sink crea automaticamente tabelle mancanti solo in Lakebase.
• I condotti a flusso lacustre non sono supportati.