Ansluta till Lakebase

Använd Structured Streaming för att skriva till Lakebase eller en extern PostgreSQL-databas med inbyggd batching, automatiska omprövningar och autentisering hanterad av arbetsytan.

När du ska använda Lakebase sink

Använd Lakebase-mottagaren för strömmande skrivningar med låg latens till Lakebase eller en extern PostgreSQL-databas. Den här sinken kräver inte att du implementerar egna foreach-funktioner för att hantera batchning, anslutningshantering och felhantering.

Vanliga användningsfall är:

• Uppdatera programdatabaser i realtid för operativa instrumentpaneler eller kundinriktade funktioner.
• Synkronisera kontinuerligt föränderliga data, till exempel aggregerade eller filtrerade direktuppspelningsresultat, till en transaktionsdatabas.
• Skriv utdata från en fråga för strukturerad direktuppspelning till en Lakebase-tabell med svarstid under sekund med realtidsläge.

Om du vill synkronisera data från Lakebase till Delta Lake-tabeller i Lakehouse, den omvända riktningen, se Lakebase Change Data Feed.

krav

• Databricks Runtime 18 LTS och uppåt.
  • Externa PostgreSQL-anslutningar kräver att du använder Databricks Runtime 19 eller senare och att du anmäler dig till förhandsversionen Custom JDBC on UC Compute.
  • Intervalldatatyper kräver att du använder Databricks Runtime 19 och uppåt.
• Klassiska beräkningstjänster med dedikerade åtkomstlägen eller standardåtkomstlägen, eller serverlösa beräkningstjänster för notebook-filer eller jobb. I serverlösa beräkningsmiljöer använder du Trigger.AvailableNow(). Se Direktuppspelning på serverlös beräkning.
• En Lakebase-databas, eller en Unity Catalog-anslutning till en extern PostgreSQL-databas.

Identifieringskrav

För alla mål rekommenderar Databricks att använda schema-, tabell-, kolumn- och primärnyckelkolumnnamn som börjar med en bokstav eller understreck och endast innehåller bokstäver, siffror och understreck. Sänkan upprätthåller dessa krav när den automatiskt skapar en Lakebase-tabell. För att använda identifierare som inte uppfyller dessa krav, skapa måltabellen innan du startar frågan.

Anslut till en databas

Lakebase-mottagaren har stöd för följande anslutningsmetoder:

Lakebase-tabeller registrerade med Unity Catalog

För Lakebase-tabeller som registrerats med Unity Catalog hanterar anslutningsappen automatiskt autentiseringsuppgifterna och använder identiteten för användaren eller tjänstens huvudnamn som kör frågan. Om tabellen inte finns skapar anslutningsappen tabellen.

Information om hur du registrerar en Lakebase-databas med Unity Catalog finns i Registrera en Lakebase-databas i Unity Catalog.

För att skriva till en Lakebase-tabell, använd .toTable() metoden med ett fullt kvalificerat tabellnamn, 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>")

Ersätt följande platshållare:

• <catalog>.<schema>.<table>: Måltabellens fullständiga kvalificerade namn. catalog är den Unity Catalog-katalog som du skapade när du registrerade Lakebase-databasen. Mer information finns i Registrera en Lakebase-databas i Unity Catalog. Om tabellen inte finns skapar anslutningsappen den.
• <primary-key-columns>: Valfritt. En kommaseparerad lista över alla kolumner i måltabellens primärnyckel, till exempel id eller user_id,event_type. Se Upsert beteende.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: En volymsökväg i Unity Catalog där frågan lagrar sin kontrollpunkt. Du kan också använda en lagrings-URI för molnobjekt. Platsen måste vara lagring som du kan skriva till, inte lokal disk, och måste vara unik för varje strömmande fråga. Detta är oberoende av måltabellen. Se Kontrollpunkter för strukturerad strömning.

För valfria konfigurationer, såsom batchsize och batchinterval, se PostgreSQL-sänkalternativen.

Lakebase-tabeller som inte har registrerats med Unity Catalog

För Lakebase-tabeller som inte har registrerats med Unity Catalog hanterar anslutningsappen automatiskt autentiseringsuppgifterna och använder identiteten för användaren eller tjänstens huvudnamn som kör frågan. Om tabellen inte finns skapar anslutningsappen tabellen.

För att skriva till en Lakebase-tabell använder du alternativen dbtable och endpoint:

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

Ersätt följande platshållare:

• <project-id>.<branch-id>.<endpoint-id>: Din Lakebase-slutpunkt. Hitta alla tre värdena i resursnamnet på menyn Hämta ID på fliken Beräkningar , som har formatet projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>. Se Beräkningsidentifierare.
• <database>: Valfritt. Namnet på den målinriktade PostgreSQL-databasen. Standardinställningen är databricks_postgres. Se Hantera databaser.
• <schema>.<table>: Måltabellen i formatet schema.table. Om du utelämnar schemat använder mottagaren schemat public. För automatisk tabellskapande, använd identifierare som börjar med en bokstav eller understreck och endast innehåller bokstäver, siffror och understreck.
• <primary-key-columns>: Valfritt. En kommaseparerad lista över alla kolumner i måltabellens primärnyckel, till exempel id eller user_id,event_type. Se Upsert beteende.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: En volymsökväg i Unity Catalog där frågan lagrar sin kontrollpunkt. Du kan också använda en lagrings-URI för molnobjekt. Platsen måste vara lagring som du kan skriva till, inte lokal disk, och måste vara unik för varje strömmande fråga. Detta är oberoende av måltabellen. Se Kontrollpunkter för strukturerad strömning.

För valfria konfigurationer, såsom batchsize och batchinterval, se PostgreSQL-sänkalternativen.

Extern PostgreSQL med Unity Catalog-inloggningsuppgifter

Använd en Unity Catalog-anslutning för att autentisera dig mot en extern PostgreSQL-databas utan att lagra inloggningsuppgifter i din kod. Måltabellen måste redan existera.

Skapa en koppling av typ POSTGRESQL, se Skapa en koppling. Användaren eller tjänstehuvudpersonen som kör frågan måste ha USE CONNECTION på anslutningen.

För att skriva till PostgreSQL-tabellen, använd databricks.connection, database, och dbtable alternativen:

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

Ersätt följande platshållare:

• <connection-name>: Namnet på Unity Catalog-kopplingen.
• <database>: Namnet på den målinriktade PostgreSQL-databasen.
• <schema>.<table>: Den befintliga måltabellen i schema.table format. Om du utelämnar schemat använder mottagaren schemat public.
• <primary-key-columns>: Valfritt. En kommaseparerad lista över alla kolumner i måltabellens primärnyckel, till exempel id eller user_id,event_type. Se Upsert beteende.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: En volymsökväg i Unity Catalog där frågan lagrar sin kontrollpunkt. Du kan också använda en lagrings-URI för molnobjekt. Platsen måste vara lagring som du kan skriva till, inte lokal disk, och måste vara unik för varje strömmande fråga. Detta är oberoende av måltabellen. Se Kontrollpunkter för strukturerad strömning.

PostgreSQL-anslutningar använder alltid TLS. Certifikatverifiering följer inställningarna på Unity Catalog-anslutningen, som du väljer när du skapar anslutningen:

• Trust server-certifikat: När det väljs använder anslutningen sslmode=require, vilket krypterar anslutningen utan att verifiera servercertifikatet.
• Användartillhandahållet servercertifikat: Tillhandahåll ett PEM-kodat servercertifikat att använda sslmode=verify-full när Trust-servercertifikat inte är valt. Om du inte tillhandahåller ett certifikat använder anslutningen sslmode=verify-full med JVM:s standardarkiv för betrodda certifikat.

Konfigurationsalternativ

Sinken ger ett fel för okända alternativ, JDBC_STREAMING_SINK_INVALID_OPTIONS.

För konfigurationsalternativen för sänkan, inklusive de gemensamma alternativen och alternativen för varje anslutningsmetod, se PostgreSQL-sänkalternativen.

Datatypsmappningar

Sinken kontrollerar att varje DataFrame-kolumn är kompatibel med sin motsvarande målkolumn innan den skriver till en befintlig Lakebase- eller extern PostgreSQL-tabell.

Följande tabell innehåller typer som stöds i Databricks Runtime 18 LTS och uppåt:

Följande tabell innehåller typer som stöds i Databricks Runtime 19 och uppåt:

Upsert-beteende

Alternativet upsertkey identifierar måltabellens primära nyckelkolumner. För en befintlig tabell måste kolumnerna i upsertkey exakt matcha tabellens primärnyckel. Om du utelämnar alternativet läser diskhon primärnyckeln från bordet. För en Lakebase-tabell som utdatasänkan skapar anger upsertkey primärnyckeln. Om du utelämnar alternativet skapar diskhon bordet utan primärnyckel.

När måltabellen har en primärnyckel utför sänket en upsert med PostgreSQL:s INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ...-syntax. När måltabellen saknar primärnyckel gör sinken infogningar. En frågas utmatningsläge påverkar inte detta beteende.

Alla primärnyckelkolumner måste finnas i DataFrame och använda jämförbara typer, såsom numeriska eller strängtyper.

Prestandaoptimering

Batchbearbetning och ryggtryck

En tömning utlöses när något av villkoren uppfylls:

• Bufferten når upp till batchsize rader, där standardvärdet är 1000.
• Buffertåldern överskrider batchinterval, vilket är standardvärdet 100 milliseconds.

När databasen inte kan hänga med i den inkommande datahastigheten sprider mottagaren tillbakatryck uppströms till källan.

Vägledning för svarstid och dataflöde:

• För arbetsbelastningar med låg latens med realtidsläge minskar du batchinterval för att garantera en kortare maximal tid före tömning. Se begrepp i realtidsläge för koncept och exempel på realtidsläge för ett kodexempel.
• För arbetsbelastningar med hög genomströmning, öka batchsize för att minska overheaden för varje transaktion.

Anslutningsbeteende

Sinken använder anslutningspoolning på executorer. Som standard använder varje aktivitet en databasanslutning.

Databricks rekommenderar att du använder standardvärdet 1 för aktiviteten för varje anslutning. Om du ökar antalet uppgifter för varje anslutning kan du orsaka anslutningskonkurrens och öka svarstiderna för anslutningar med högt dataflöde.

Om du vill konfigurera förhållandet mellan aktiviteter och anslutningar anger du spark.databricks.sql.streaming.jdbc.tasksPerConnection Spark-konfigurationen. Om måldatabasen har en låg anslutningsgräns kan du minska antalet shuffle-partitioner eller öka spark.databricks.sql.streaming.jdbc.tasksPerConnection.

Sinken försöker automatiskt igen vid tillfälliga fel i JDBC, inklusive anslutningsfel, dödlägen och frekvensbegränsning. Om sinken har förbrukat alla återförsök, misslyckas frågan.

Utlösare och utdatalägen som stöds

Triggers

Denna tabell visar stöd för Structured Streaming-triggertyper på klassiska och serverlösa beräkningar:

Utdatalägen

Den här tabellen visar stöd för utdatalägen för strukturerad direktuppspelning:

Begränsningar

• För en extern PostgreSQL-databas ansluten via en Unity Catalog-anslutning måste måltabellen redan existera. Sänkan skapar automatiskt saknade tabeller endast i Lakebase.
• Lakeflow-pipelines stöds inte.