Csatlakozás a Lakebase-hez

A strukturált streamelés használatával beépített kötegeléssel, automatikus újrapróbálkozással és munkaterület által felügyelt hitelesítéssel írhat a Lakebase-be.

Mikor használja a Lakebase sinket?

Használja a Lakebase-fogadót alacsony késésű streamelési írásokhoz a Lakebase-be. Ez a fogadó nem követeli meg, hogy egyéni foreachBatch függvényeket implementáljon a kötegelés, a kapcsolatkezelés és a hibakezelés kezeléséhez.

Gyakori használati esetek a következők:

• Valós időben frissítheti az alkalmazás-adatbázisokat az operatív irányítópultok vagy az ügyféloldali funkciók számára.
• Szinkronizálja a folyamatosan változó adatokat, például az összesített vagy szűrt streamelési eredményeket egy tranzakciós adatbázisba.
• Írja be egy strukturált streamelési lekérdezés kimenetét egy Lakebase-táblába, amely másodperc alatti késéssel rendelkezik valós idejű módban.

A Lakebase-ből a Lakehouse Delta Lake-tábláiba történő adatszinkronizáláshoz, a fordított iránnyal kapcsolatban lásd: Lakebase Change Data Feed.

Követelmények

• Databricks Runtime 18 vagy újabb
• Klasszikus számítás dedikált vagy standard hozzáférési módokkal.
• Egy Lakebase-adatbázis

Kapcsolódás adatbázishoz

A Lakebase-fogadó a következő csatlakozási módszereket támogatja:

A Unity Katalógusban regisztrált Lakebase-táblák

A Unity Catalogban regisztrált Lakebase-táblák esetében az összekötő automatikusan kezeli a hitelesítő adatokat, és a lekérdezést futtató felhasználó vagy szolgáltatásnév identitását használja. Ha a tábla nem létezik, az összekötő létrehozza a táblát.

Ha lakebase-adatbázist szeretne regisztrálni a Unity Catalogban, olvassa el a Lakebase-adatbázis regisztrálása a Unity Catalogban című témakört.

Ha Egy Lakebase-táblába szeretne írni, használja a .toTable() metódust egy teljesen minősített táblanévvel. catalog.schema.table Az alábbi példa a szükséges beállításokat, valamint az opcionális upsertkey lehetőséget mutatja be:

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

Cserélje ki a következő kitöltendő elemeket.

• <catalog>.<schema>.<table>: A céltábla teljes neve. Ez catalog a Lakebase-adatbázis regisztrálásakor létrehozott Unity Catalog-katalógus. Lásd: Lakebase-adatbázis regisztrálása a Unity Katalógusban. Ha a tábla nem létezik, az összekötő létrehozza.
• <primary-key-column>:Szabadon választható. Az upsert kulcsot alkotó oszlopok vesszővel tagolt listája, például iduser_id,event_typevagy . Ha kihagyja a upsertkey elemet, a fogadóoldal a céltábla elsődleges kulcsa alapján kikövetkezteti a kulcsot. Lásd a Upsert viselkedését.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Egy Unity-katalógus kötetútvonala, ahol a lekérdezés tárolja az ellenőrzőpontot. Felhőalapú objektumtárolási URI-t is használhat. A helynek olyan tárolónak kell lennie, amelybe írhat, nem pedig helyi lemezre, és egyedinek kell lennie az egyes streamelési lekérdezések esetében. Ez független a céltáblától. Lásd: Strukturált streamelési ellenőrzőpontok.

Az opcionális konfigurációkkal, például batchsize és batchinterval kapcsolatban lásd: Konfigurációs lehetőségek.

A Unity Catalogban nem regisztrált Lakebase-táblák

A Unity Catalogban nem regisztrált Lakebase-táblák esetében az összekötő automatikusan kezeli a hitelesítő adatokat, és a lekérdezést futtató felhasználó vagy szolgáltatásnév identitását használja. Ha a tábla nem létezik, az összekötő létrehozza a táblát.

Ha egy Lakebase-táblába szeretne írni, használja a endpoint és dbtable beállításokat. Az alábbi példa a választható database és upsertkey lehetőségeket is tartalmazza:

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

Cserélje ki a következő kitöltendő elemeket.

• <project-id>.<branch-id>.<endpoint-id>: A Lakebase-végpont. Keresse meg mindhárom értéket az erőforrásnévben a Számítások lap Get ID menüjében, amelynek formátumaprojects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>. Lásd : Számítási azonosítók.
• <database>:Szabadon választható. A cél Postgres-adatbázis neve. Alapértelmezett érték: databricks_postgres. Lásd: Adatbázisok kezelése.
• <schema>.<table>: A(z) schema.table formátumú céltábla. Ha kihagyja a sémát, a fogadó a public sémát használja. Használjon egyszerű azonosítókat, amelyek betűvel vagy aláhúzásjellel kezdődnek, és csak betűket, számokat és aláhúzásjeleket tartalmaznak; Az idézett azonosítók és speciális karakterek, például kötőjelek nem támogatottak.
• <primary-key-column>:Szabadon választható. Az upsert kulcsot alkotó oszlopok vesszővel tagolt listája, például iduser_id,event_typevagy . Ha kihagyja a upsertkey elemet, a fogadóoldal a céltábla elsődleges kulcsa alapján kikövetkezteti a kulcsot. Lásd a Upsert viselkedését.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Egy Unity-katalógus kötetútvonala, ahol a lekérdezés tárolja az ellenőrzőpontot. Felhőalapú objektumtárolási URI-t is használhat. A helynek olyan tárolónak kell lennie, amelybe írhat, nem pedig helyi lemezre, és egyedinek kell lennie az egyes streamelési lekérdezések esetében. Ez független a céltáblától. Lásd: Strukturált streamelési ellenőrzőpontok.

Az opcionális konfigurációkkal, például batchsize és batchinterval kapcsolatban lásd: Konfigurációs lehetőségek.

Konfigurációs beállítások

A nyelő hibát jelez az ismeretlen beállítások esetén, JDBC_STREAMING_SINK_INVALID_OPTIONS.

Az alábbi beállítások az összes kapcsolati módszerre vonatkoznak:

A Unity Catalogban nem regisztrált Lakebase-táblák

A következő beállítások érvényesek, ha olyan Lakebase-táblához csatlakozik, amely nincs regisztrálva a Unity Catalogban:

Upsert működése

Ha léteznek upsert kulcsok, amelyeket vagy a upsertkey elemmel adnak meg, vagy amelyeket a nyelő a tábla elsődleges kulcsaiból következtet ki, a nyelő a PostgreSQL INSERT INTO ... ON CONFLICT (<upsert_key>) DO UPDATE SET ... szintaxisával hajt végre upsert műveletet a táblán.

Ha nincsenek upsert kulcsok, a nyelő insert műveleteket végez. A lekérdezés kimeneti módja nincs hatással az upsert- vagy insert-művelet viselkedésére.

Az upsertkey oszlopoknak a következőnek kell lenniük:

• Legyen a DataFrame-oszlopok nem üres részhalmaza.
• Pontosan egyezzen meg a céltáblával PRIMARY KEY. Ha a megadott oszlopok nem egyeznek az elsődleges kulccsal, a lekérdezés meghiúsul.
• Legyen összehasonlítható típusok, például numerikus vagy sztringtípusok. Az adatbázis-holtpontok párhuzamos írások során történő elkerülése érdekében a nyelő az egyes kötegeken belül az upsert kulcs szerint rendezi a sorokat. Az Upsert-kulcsok nem támogatják az összetett vagy strukturált típusokat.

Az oszlopneveket a rendszer automatikusan a PostgreSQL alapértelmezett kettős idézőjele közé teszi ", ami lehetővé teszi a fenntartott kulcsszavak és a vegyes kis- és nagybetűs nevek kezelését.

A táblázat- és sémaneveknek olyan egyszerű azonosítókat kell használniuk, amelyek betűvel vagy aláhúzásjellel kezdődnek, és csak betűket, számokat és aláhúzásjeleket tartalmaznak. A fogadóoldal nem támogatja az idézőjelek közé tett azonosítókat és a speciális karaktereket, például a kötőjeleket a tábla- vagy sémanevekben.

Teljesítmény finomhangolása

Kötegelés és visszanyomás

A kiürítés akkor aktiválódik, ha valamelyik feltétel teljesül:

• A puffer mérete eléri a batchsize sort, amelynek alapértelmezett értéke 1000.
• A puffer kora meghaladja a(z) batchinterval értéket, amelynek alapértelmezett értéke 100 milliseconds.

Ha az adatbázis nem képes lépést tartani a bejövő adatok sebességével, a nyelő backpressure-t továbbít visszafelé a forrás felé.

Késleltetésre és áteresztőképességre vonatkozó útmutató:

• Valós idejű móddal működő, alacsony késleltetésű munkaterhelések esetén csökkentse a(z) batchinterval értékét, hogy garantálja a kiírás előtti rövidebb maximális időt. A strukturált streamelés valós idejű módjával kapcsolatos fogalmakat és valós idejű módú példákat talál egy kód példájához.
• A nagy átviteli sebességű számítási feladatok esetében növelje batchsize az egyes tranzakciók terhelésének csökkentéséhez.

Kapcsolat viselkedése

A nyelő kapcsolatpoolozást használ a végrehajtó folyamatokon. Alapértelmezés szerint minden tevékenység egy adatbázis-kapcsolatot használ.

A Databricks azt javasolja, hogy minden kapcsolathoz használja a tevékenység alapértelmezett értékét 1 . Ha növeli az egyes kapcsolatokhoz tartozó feladatok számát, előfordulhat, hogy kapcsolati versengéseket okoz, és növeli a késéseket a nagy átviteli sebességű kapcsolatok esetében.

A tevékenységek kapcsolatokhoz viszonyított arányának konfigurálásához állítsa be a Spark-konfigurációt spark.databricks.sql.streaming.jdbc.tasksPerConnection . Ha a céladatbázisnak alacsony a kapcsolati korlátja, csökkentse az shuffle partíciók számát, vagy növelje spark.databricks.sql.streaming.jdbc.tasksPerConnection.

A fogadó automatikusan újrapróbálkozza az átmeneti JDBC-hibákat, beleértve a csatlakozási hibákat, a holtpontokat és a sebességkorlátozást. Ha a nyelőnél kimerül az összes újrapróbálkozási lehetőség, a lekérdezés meghiúsul.

Támogatott eseményindítók és kimeneti módok

Kiváltó okok

Ez a táblázat a strukturált streamelési eseményindító-típusok támogatását mutatja be:

Kimeneti módok

Ez a táblázat a strukturált streamelési kimeneti módok támogatását mutatja be:

korlátozások

• A kiszolgáló nélküli számítási és Lakeflow-folyamatok nem támogatottak.
• Írási célként csak a Lakebase támogatott. A külső PostgreSQL-kompatibilis adatbázisok nem támogatottak.