Připojení k Lakebase

Použijte Structured Streaming k zápisu do Lakebase nebo externí databáze PostgreSQL s vestavěným dávkováním, automatickými opakovaními a autentizací spravovanou pracovním prostorem.

Kdy použít jímku Lakebase

Používejte jímku Lakebase pro streamované zápisy s nízkou latencí do Lakebase nebo externí databáze PostgreSQL. Tento cíl nevyžaduje, abyste implementovali vlastní funkce foreach pro dávkové zpracování, správu připojení a zpracování chyb.

Mezi běžné případy použití patří:

Aktualizujte aplikační databáze v reálném čase pro provozní řídicí panely nebo funkce určené pro zákazníky.
Synchronizovat průběžně se měnící data, jako jsou agregované nebo filtrované výsledky streamování, do transakční databáze.
Zapište výstup dotazu strukturovaného streamování do tabulky Lakebase s latencí podsekundy pomocí režimu v reálném čase.

Pro synchronizaci dat z Lakebase do tabulek Delta Lake v Lakehouse, tedy v opačném směru, viz Kanál změn dat Lakebase.

Požadavky

Databricks Runtime 18 LTS a vyšší.
- Externí připojení PostgreSQL vyžadují použití Databricks Runtime 19 a novější a přihlášení k Custom JDBC na UC Compute preview.
- Intervalové datové typy vyžadují použití Databricks Runtime 19 a vyšší.
- Strategie slučování pro upserty vyžadují použití Databricks Runtime ve verzi 19 a novější.
Klasické výpočetní prostředky s vyhrazenými nebo standardními režimy přístupu, nebo bezserverové výpočetní prostředky pro poznámkové bloky či úlohy. Pro serverless výpočty použijte Trigger.AvailableNow(). Viz Streamování na výpočetních prostředcích bez serveru.
Databáze Lakebase nebo připojení Unity Catalog k externí databázi PostgreSQL.

Požadavky na identifikátor

Pro všechny cílové systémy Databricks doporučuje používat názvy schémat, tabulek, sloupců a sloupců primárních klíčů, které začínají písmenem nebo podtržítkem a obsahují pouze písmena, číslice a podtržítka. Sink tyto požadavky vynucuje, když automaticky vytváří tabulku Lakebase. Pro použití identifikátorů, které tyto požadavky nespĺnějí, vytvořte cílovou tabulku před zahájením dotazu.

Připojit k databázi

Jímka Lakebase podporuje následující metody připojení:

Tabulky Lakebase zaregistrované v katalogu Unity

V případě tabulek Lakebase zaregistrovaných v Unity Catalog konektor automaticky spravuje přihlašovací údaje a používá identitu uživatele nebo instančního objektu služby, který dotaz spouští. Pokud tabulka neexistuje, konektor vytvoří tabulku.

Pokud chcete zaregistrovat databázi Lakebase v katalogu Unity, přečtěte si téma Registrace databáze Lakebase v katalogu Unity.

Pro zápis do tabulky Lakebase použijte metodu .toTable() s plně kvalifikovaným názvem tabulky: 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>")

Nahraďte následující zástupné symboly:

<catalog>.<schema>.<table>: Plně kvalifikovaný název cílové tabulky. Jedná se catalog o katalog Unity, který jste vytvořili při registraci databáze Lakebase, viz Registrace databáze Lakebase v Katalogu Unity. Pokud tabulka neexistuje, konektor ji vytvoří.
<primary-key-columns>: Volitelné. Čárkami oddělený seznam všech sloupců v primárním klíči cílové tabulky, například id nebo user_id,event_type. Viz chování Upsert.
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Cesta ke svazku katalogu Unity, kde dotaz ukládá svůj kontrolní bod. Můžete také použít identifikátor URI cloudového úložiště objektů. Umístění musí být úložiště, do kterého můžete zapisovat, nikoli na místní disk, a musí být jedinečné pro každý dotaz streamování. To je nezávislé na cílové tabulce. Viz kontrolní body strukturovaného streamování.

Pro volitelné konfigurace, jako jsou batchsize a batchinterval, viz možnosti cíle PostgreSQL.

Tabulky Lakebase nejsou zaregistrované v katalogu Unity

U tabulek Lakebase, které nejsou zaregistrované ve službě Unity Catalog, konektor automaticky spravuje přihlašovací údaje a používá identitu uživatele nebo instančního objektu, který dotaz spouští. Pokud tabulka neexistuje, konektor vytvoří tabulku.

Pro zápis do tabulky Lakebase použijte možnosti endpoint a 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()

Nahraďte následující zástupné symboly:

<project-id>.<branch-id>.<endpoint-id>: Váš koncový bod Lakebase. Všechny tři hodnoty v názvu prostředku najdete v nabídce Získat ID na kartě Výpočty , která má formát projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>. Viz Identifikátory výpočetních prostředků.
<database>: Volitelné. Název cílové databáze PostgreSQL. Výchozí hodnota je databricks_postgres. Viz Správa databází.
<schema>.<table>: Cílová tabulka ve schema.table formátu. Pokud schéma vynecháte, jímka použije public schéma. Pro automatické vytváření tabulek používejte identifikátory začínající písmenem nebo podtržítkem a obsahující pouze písmena, čísla a podtržítka.
<primary-key-columns>: Volitelné. Čárkami oddělený seznam všech sloupců v primárním klíči cílové tabulky, například id nebo user_id,event_type. Viz chování Upsert.
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Cesta ke svazku katalogu Unity, kde dotaz ukládá svůj kontrolní bod. Můžete také použít identifikátor URI cloudového úložiště objektů. Umístění musí být úložiště, do kterého můžete zapisovat, nikoli na místní disk, a musí být jedinečné pro každý dotaz streamování. To je nezávislé na cílové tabulce. Viz kontrolní body strukturovaného streamování.

Pro volitelné konfigurace, jako jsou batchsize a batchinterval, viz možnosti cíle PostgreSQL.

Externí PostgreSQL s přihlašovacími údaji Unity Catalog

Important

Tato funkce je ve verzi Public Preview. Správci pracovního prostoru mohou ovládat přístup k Custom JDBC na UC Compute ze stránky Previews . Viz Manage Azure Databricks preview.

Použijte připojení Unity Catalog k autentizaci k externí databázi PostgreSQL bez ukládání přihlašovacích údajů do kódu. Cílová tabulka už musí existovat.

Vytvořte spojení typu POSTGRESQL, viz Vytvořit spojení. Uživatel nebo princip služby, který dotaz provádí, musí mít USE CONNECTION na spojení.

Pro zápis do tabulky PostgreSQL použijte databricks.connection, database, a dbtable možnosti:

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

Nahraďte následující zástupné symboly:

<connection-name>: Název připojení ke službě Unity Catalog.
<database>: Název cílové databáze PostgreSQL.
<schema>.<table>: Stávající cílová tabulka ve formátu schema.table . Pokud schéma vynecháte, jímka použije public schéma.
<primary-key-columns>: Volitelné. Čárkami oddělený seznam všech sloupců v primárním klíči cílové tabulky, například id nebo user_id,event_type. Viz chování Upsert.
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Cesta ke svazku katalogu Unity, kde dotaz ukládá svůj kontrolní bod. Můžete také použít identifikátor URI cloudového úložiště objektů. Umístění musí být úložiště, do kterého můžete zapisovat, nikoli na místní disk, a musí být jedinečné pro každý dotaz streamování. To je nezávislé na cílové tabulce. Viz kontrolní body strukturovaného streamování.

PostgreSQL připojení vždy používá TLS. Ověření certifikátu probíhá podle nastavení připojení v Unity Catalog, která zvolíte při vytváření připojení:

Trust server certificate: Po výběru spojení použije sslmode=require, což zašifruje spojení bez ověření serverového certifikátu.
Uživatelem poskytnutý serverový certifikát: Poskytnout PEM kódovaný serverový certifikát pro použití sslmode=verify-full , když není vybrán certifikát Trust serveru . Pokud neposkytnete certifikát, připojení používá sslmode=verify-full s výchozím úložištěm důvěry JVM.

Možnosti konfigurace

Sink vrátí chybu při nerozpoznaných volbách, JDBC_STREAMING_SINK_INVALID_OPTIONS.

Pro možnosti konfigurace sinku, včetně běžných možností a možností pro každou metodu připojení, viz PostgreSQL sink options.

Mapování datových typů

Jímka před zápisem do existující tabulky Lakebase nebo externí tabulky PostgreSQL kontroluje, zda je každý sloupec objektu DataFrame kompatibilní s odpovídajícím cílovým sloupcem.

Následující tabulka obsahuje typy podporované v Databricks Runtime 18 LTS a vyšších:

Typ Spark	Automaticky vytvářený typ tabulky Lakebase	Kompatibilní typy v existujících tabulkách PostgreSQL
`ByteType`, `ShortType`	`smallint`	`smallint`
`IntegerType`	`integer`	`integer`
`LongType`	`bigint`	`bigint`
`FloatType`	`real`	`real`
`DoubleType`	`double precision`	`double precision`
`DecimalType`	`numeric`	`numeric`
`StringType`	`text`	`varchar`, `text`
`VarcharType(n)`	`varchar(n)`	`varchar`, `text`
`CharType(n)`	`char(n)`	`char`
`BinaryType`	`bytea`	`bytea`
`BooleanType`	`boolean`	`boolean`
`TimestampType`	`timestamptz`	`timestamptz`
`TimestampNTZType`	`timestamp`	`timestamp`
`DateType`	`date`	`date`
`ArrayType`, `MapType`, `StructType`, , `VariantTypeNullType`	`jsonb`	`json`, `jsonb`

Následující tabulka obsahuje typy podporované v Databricks Runtime 19 a výše:

Typ Spark	Automaticky vytvářený typ tabulky Lakebase	Kompatibilní typy v existujících tabulkách PostgreSQL
`DayTimeIntervalType`, `YearMonthIntervalType`	`interval`	`interval`

Chování operace upsert

Možnost upsertkey identifikuje sloupce primárního klíče cílové tabulky. U existující tabulky musí sloupce v upsertkey přesně odpovídat primárnímu klíči tabulky. Pokud tuto možnost vynecháte, dřez přečte hlavní klíč ze stolu. Pro tabulku Lakebase, kterou sink vytváří, definuje upsertkey primární klíč. Pokud tuto možnost vynecháte, dřez vytvoří stůl bez primárního klíče.

Když má cílová tabulka primární klíč, sink se vyrovná se syntaxí INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ... PostgreSQL. Když cílová tabulka nemá primární klíč, dřez provádí vložení. Výstupní režim dotazu nemá na toto chování žádný vliv.

Všechny sloupce primárních klíčů musí být přítomny v DataFrame a používat srovnatelné typy, například číselné nebo řetězcové typy.

Pro kontrolu, jak upserty aktualizují sloupce jiné než primární klíč, viz Merge strategies for upserts.

Strategie slučování pro upserty

V Databricks Runtime 19 a vyšších verzích můžete použít strategie sloučení k ovládání toho, jak sink aktualizuje sloupce, které nejsou součástí primárního klíče, když mají řádky stejný primární klíč. Například můžete zachovat největší časové razítko nebo aktualizovat sloupec pouze tehdy, když má příchozí událost novější hodnotu pořadí.

Strategie sloučení se vztahují pouze na upserty. Pokud cílová tabulka nemá primární klíč, nastavení upsert.mergeStrategies nebo upsert.defaultMergeStrategy způsobí chybu. Strategie můžete přiřadit pouze sloupcům, které nejsou primárním klíčem. Nastavte upsert.mergeStrategies na seznam oddělený čárkami ve formátu column:strategy. Nastavte upsert.defaultMergeStrategy, aby se nakonfigurovala výchozí hodnota pro sloupce bez explicitních strategií. upsert.defaultMergeStrategy nepřijímá latestBy ani earliestBy. Ve výchozím nastavení sink používá strategii latest.

Podporované strategie sloučení

Strategy	Behavior
`latest`	Použijte nejnovější hodnotu podle pořadí doručení. Toto je výchozí hodnota.
`earliest`	Použijte nejdřívější hodnotu podle pořadí příjezdu.
`latestNonNull`	Použijte nejnovější nenulovou hodnotu podle pořadí příchodu.
`earliestNonNull`	Použijte nejstarší nenulovou hodnotu podle pořadí příchodu.
`greatest`	Použijte vyšší z existující a příchozí hodnoty. Nulové hodnoty jsou ignorovány, takže nulová příchozí hodnota nenahrazuje existující hodnotu.
`least`	Použijte nižší z existující a příchozí hodnoty. Nulové hodnoty jsou ignorovány, takže nulová příchozí hodnota nenahrazuje existující hodnotu.
`latestBy(column)`	Použijte hodnotu z řádku s největší hodnotou ve specifikovaném sloupci pořadí. Podívejte se, jak `latestBy` a `earliestBy` pracují s remízami a hodnotami null.
`earliestBy(column)`	Použijte hodnotu z řádku s nejmenší hodnotou v zadaném sloupci pro řazení. Podívejte se na Jak `latestBy` a `earliestBy` zpracovávají shody a hodnoty null.

Jak `latestBy` a `earliestBy` pracují s vazbami a nulovými hodnotami

latestBy(column) a earliestBy(column) vybírají hodnotu porovnáním příchozí hodnoty řazení s tou, která je již uložena v tabulce. Aby byla tato uložená hodnota přesná, musí sloupec pořadí používat strategii párování, buď v upsert.mergeStrategies, nebo prostřednictvím upsert.defaultMergeStrategy:

Pro latestBy(column) musí sloupec pořadí použít greatest.
Pro earliestBy(column) musí sloupec pořadí použít least.

Například is_success:latestBy(last_updated_at) vyžaduje last_updated_at:greatest.

Když jsou obě pořadové hodnoty nenulové, latestBy(column) aktualizuje sloupec pouze tehdy, když je příchozí pořadová hodnota přísně větší (>) než stávající pořadová hodnota. U earliestBy(column) musí být přísně menší (<). Pro obě strategie, pokud jsou hodnoty pořadí stejné, sink zachovává stávající hodnotu sloupce. V rámci jedné dávky, pokud má více řádků stejný primární klíč a hodnotu pořadí, sink si ponechá první řádek, který přijme.

Například pokud is_success používá latestBy(last_updated_at) a příchozí řádek má stejnou last_updated_at hodnotu jako stávající řádek, sink si zachová stávající is_success, i když se příchozí hodnota liší.

U obou strategií mají přednost hodnoty ne-nulového uspořádání před hodnotami nulového uspořádání. Příchozí řádek s nenulovou hodnotou pořadí nahrazuje existující řádek, jehož hodnota pořadí je nulová. Příchozí řádek s nulovou hodnotou pořadí nenahrazuje existující hodnotu, včetně případů, kdy je stávající pořadací hodnota také nulová.

Příklad: Vlastní logika upsertu modelu

Uvažujme tabulku se sloupci id, last_updated_at, creation_date, a is_success. Možná byste chtěli vyjádřit následující logiku:

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;

Pro modelování této logiky použijte následující strategie sloučení:

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

Ladění výkonu

Dávkování a zpětný tlak

Při splnění některé z podmínek se aktivuje vyprázdnění:

Vyrovnávací paměť dosáhne velikosti batchsize řádků, přičemž výchozí hodnota je 1000.
Stáří vyrovnávací paměti překračuje batchinterval, což je výchozí hodnota 100 milliseconds.

Pokud databáze nedokáže zpracovávat příchozí data požadovanou rychlostí, cílová komponenta propaguje zpětný tlak proti proudu ke zdroji.

Pokyny k latenci a propustnosti:

U úloh s nízkou latencí v režimu reálného času snižte batchinterval, abyste zajistili kratší maximální dobu před vyprázdněním. Viz koncepty režimu v reálném čase pro koncepty a příklady režimu v reálném čase pro příklad kódu.
U úloh s vysokou propustností zvyšte batchsize, abyste snížili režii u každé transakce.

Chování připojení

Jímka používá sdružování připojení u exekutorů. Ve výchozím nastavení používá každý úkol jedno připojení k databázi.

Databricks doporučuje, abyste pro každé připojení použili výchozí hodnotu 1 úkolu. Pokud zvýšíte počet úloh pro každé připojení, můžete způsobit kolize připojení a zvýšit latenci u připojení s vysokou propustností.

Pokud chcete nakonfigurovat poměr úloh k připojením, nastavte konfiguraci Sparku spark.databricks.sql.streaming.jdbc.tasksPerConnection . Pokud má cílová databáze nízký limit připojení, snižte počet oddílů pro přeskupení nebo zvyšte spark.databricks.sql.streaming.jdbc.tasksPerConnection.

Jímka automaticky opakuje přechodné chyby JDBC, včetně selhání připojení, zablokování a omezování rychlosti. Pokud jímka vyčerpá všechny pokusy o opakování, dotaz selže.

Podporované aktivační události a výstupní režimy

Triggers

Tato tabulka ukazuje podporu typů spouštěčů strukturovaného streamování na klasickém a serverless výpočetním systému:

Trigger	Klasické výpočetní prostředky	Bezserverové výpočty (notebooky a úlohy)
`RealTime`	Ano	No
`ProcessingTime`	Ano	No
`AvailableNow`	Ano	Ano
`Once`	Yes. Deprecated. Použijte `AvailableNow`.	Yes. Deprecated. Použijte `AvailableNow`.

Výstupní režimy

Tato tabulka ukazuje podporu pro režimy výstupu strukturovaného streamování:

Výstupní režim	Podporováno
`update`	Ano
`append`	Yes. Chování je identické s `update`. Dotaz provede operaci upsert, pokud má cílová tabulka primární klíč; jinak provede vložení. Viz chování Upsert.
`complete`	No

omezení

Pro externí PostgreSQL databázi připojenou přes Unity Catalog musí cílová tabulka již existovat. Dřez automaticky vytváří chybějící tabulky pouze v Lakebase.
Potrubí Lakeflow není podporováno.

Váš názor

Byla tato stránka užitečná?

Poslední aktualizace v 2026-11-04

Připojení k Lakebase

Kdy použít jímku Lakebase

Požadavky

Požadavky na identifikátor

Připojit k databázi

Tabulky Lakebase zaregistrované v katalogu Unity

Python

Scala

Tabulky Lakebase nejsou zaregistrované v katalogu Unity

Python

Scala

Externí PostgreSQL s přihlašovacími údaji Unity Catalog

Python

Scala

Možnosti konfigurace

Mapování datových typů

Chování operace upsert

Strategie slučování pro upserty

Podporované strategie sloučení

Jak latestBy a earliestBy pracují s vazbami a nulovými hodnotami

Příklad: Vlastní logika upsertu modelu

Python

Scala

Ladění výkonu

Dávkování a zpětný tlak

Chování připojení

Podporované aktivační události a výstupní režimy

Triggers

Výstupní režimy

omezení

Váš názor

Další materiály

Jak `latestBy` a `earliestBy` pracují s vazbami a nulovými hodnotami