Připojení k Lakebase

Důležité

Tato funkce je ve verzi Public Preview.

Pomocí strukturovaného streamování zapisujte do Lakebase s integrovaným dávkováním, automatickými opakovanými pokusy a ověřováním spravovaným pracovním prostorem.

Kdy použít jímku Lakebase

Použijte sink Lakebase pro streamované zápisy s nízkou latencí do Lakebase. Tento cíl nevyžaduje, abyste implementovali vlastní funkce foreachBatch 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 a novější
  • Klasické výpočetní prostředky s vyhrazenými nebo standardními režimy přístupu.
  • Databáze Lakebase

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.

K zápisu do tabulky Lakebase použijte .toTable() metodu s plně kvalifikovaným názvem tabulky . catalog.schema.table Následující příklad ukazuje požadované možnosti a volitelnou upsertkey možnost:

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

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-column>: Volitelné. Seznam sloupců oddělených čárkami, které tvoří klíč pro operaci upsert, například id nebo user_id,event_type. Pokud tento klíč vynecháte upsertkey, jímka odvodí klíč z primárního klíče cílové tabulky. Viz Chování upsertu.
  • /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í.

Volitelné konfigurace, například batchsize a batchinterval, viz Možnosti konfigurace.

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. Následující příklad obsahuje také volitelné database a upsertkey možnosti:

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

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 Postgres. 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. Používejte jednoduché identifikátory, které začínají písmenem nebo podtržítkem a obsahují pouze písmena, číslice a podtržítka; identifikátory v uvozovkách ani speciální znaky, jako jsou spojovníky, nejsou podporovány.
  • <primary-key-column>: Volitelné. Seznam sloupců oddělených čárkami, které tvoří klíč pro operaci upsert, například id nebo user_id,event_type. Pokud tento klíč vynecháte upsertkey, jímka odvodí klíč z primárního klíče cílové tabulky. Viz Chování upsertu.
  • /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í.

Volitelné konfigurace, například batchsize a batchinterval, viz Možnosti konfigurace.

Možnosti konfigurace

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

Následující možnosti platí pro všechny metody připojení:

Key Výchozí Description
batchinterval 100 milliseconds Optional. Maximální doba uložení řádků do vyrovnávací paměti před vyprázdněním. Například: "50 milliseconds".
batchsize 1000 Optional. Maximální počet řádků pro každou transakci databáze.
checkpointLocation None Required. Cesta k adresáři kontrolních bodů, například ke svazku katalogu Unity (/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>). Každý dotaz musí být jedinečný. Viz kontrolní body strukturovaného streamování.
upsertkey None Optional. Seznam názvů sloupců oddělený čárkou, které tvoří klíč upsertu. Například "id" nebo "user_id,event_type". Pokud zadáte upsertkey, musí sloupce odpovídat primárnímu klíči tabulky nebo dotaz selže. Pokud ho vynecháte, jímka použije primární klíč automaticky. Další informace naleznete v tématu chování operace Upsert.

Tabulky Lakebase nejsou zaregistrované v katalogu Unity

Následující možnosti platí pro připojení k tabulce Lakebase, která není zaregistrovaná v katalogu Unity:

Key Výchozí Description
database databricks_postgres Optional. Název cílové databáze PostgreSQL.
dbtable None Required. Název cílové tabulky ve schema.table formátu. Pokud nezadáte schéma, výchozí hodnota schématu je public. Používejte jednoduché identifikátory, které začínají písmenem nebo podtržítkem a obsahují jenom písmena, číslice a podtržítka. Neuvozujte názvy tabulek nebo schémat; Uvozované identifikátory a názvy se speciálními znaky, jako jsou spojovníky, nejsou podporovány.
endpoint None Required. Koncový bod Lakebase ve formátu project_id.branch_id nebo project_id.branch_id.endpoint_id. Tato endpoint_id možnost je nepovinná. Pokud ji vynecháte a větev obsahuje jeden koncový bod pro čtení i zápis, jímka ve výchozím nastavení vybere tento koncový bod.

Chování upsertu

Pokud existují klíče pro upsert, ať už zadané pomocí upsertkey nebo odvozené jímkou z primárních klíčů tabulky, jímka provádí upsert do tabulky pomocí syntaxe PostgreSQL INSERT INTO ... ON CONFLICT (<upsert_key>) DO UPDATE SET ....

Pokud neexistují žádné upsertové klíče, jímka provede vložení. Výstupní režim dotazu nemá vliv na chování operací upsert ani insert.

Sloupce upsertkey musí:

  • Musí být neprázdnou podmnožinou sloupců objektu DataFrame.
  • Přesně odpovídejte cílové tabulce PRIMARY KEY. Pokud zadané sloupce neodpovídají primárnímu klíči, dotaz selže.
  • Musí jít o srovnatelné typy, například číselné nebo řetězcové typy. Aby se zabránilo zablokování databáze během souběžných zápisů, jímka seřadí řádky podle klíče upsertu v každé dávce. Upsertové klíče nepodporují složité typy ani typy struktur.

Názvy sloupců se automaticky uzavírají do výchozích uvozovek PostgreSQL, dvojitých uvozovek ", což umožňuje správně pracovat s vyhrazenými klíčovými slovy a názvy kombinujícími malá a velká písmena.

Názvy tabulek a schémat musí používat jednoduché identifikátory, které začínají písmenem nebo podtržítkem a obsahují pouze písmena, číslice a podtržítka. Jímka nepodporuje uvozované identifikátory ani speciální znaky, jako jsou spojovníky, v názvech tabulek nebo schémat.

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:

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ů triggerů strukturovaného streamování:

Trigger Podporováno
realTime Ano
ProcessingTime Ano
AvailableNow Ano
Once Ano

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í upsertu.
complete No

omezení

  • Bezserverové výpočetní prostředky a kanály Lakeflow se nepodporují.
  • Jako cíl zápisu se podporuje jenom Lakebase. Externí databáze kompatibilní s PostgreSQL se nepodporují.
  • Last updated on