Hubungkan ke Lakebase

Gunakan Structured Streaming untuk menulis ke Lakebase atau database PostgreSQL eksternal dengan pemrosesan batch bawaan, percobaan ulang otomatis, dan autentikasi yang dikelola oleh workspace.

Kapan menggunakan sink Lakebase

Gunakan sink Lakebase untuk penulisan streaming dengan latensi rendah ke Lakebase atau database PostgreSQL eksternal. Sink ini tidak mengharuskan Anda menerapkan fungsi kustom foreach untuk menangani batching, manajemen koneksi, dan penanganan kesalahan.

Kasus penggunaan umum meliputi:

  • Perbarui database aplikasi secara real time untuk dasbor operasional atau fitur yang menghadap pelanggan.
  • Sinkronkan data yang terus berubah, seperti hasil streaming yang diagregasi atau difilter, ke dalam basis data transaksional.
  • Tulis output kueri Streaming Terstruktur ke dalam tabel Lakebase dengan latensi sub-detik menggunakan mode real time.

Untuk menyinkronkan data dari tabel Lakebase ke Delta Lake di Lakehouse, arah sebaliknya, lihat Umpan Data Perubahan Lakebase.

Persyaratan

  • Databricks Runtime 18 LTS ke atas.
    • Koneksi PostgreSQL eksternal mengharuskan Anda menggunakan Databricks Runtime 19 atau yang lebih baru dan ikut serta dalam pratinjau Custom JDBC pada UC Compute.
    • Tipe data interval mengharuskan Anda menggunakan Databricks Runtime 19 ke atas.
    • Strategi penggabungan untuk upsert mengharuskan Anda menggunakan Databricks Runtime 19 ke atas.
  • Komputasi klasik dengan mode akses khusus atau standar, atau komputasi serverless untuk notebook atau pekerjaan. Pada komputasi serverless, gunakan Trigger.AvailableNow(). Lihat Streaming pada komputasi tanpa server.
  • Database Lakebase, atau koneksi Unity Catalog ke database PostgreSQL eksternal.

Persyaratan pengidentifikasi

Untuk semua target, Databricks merekomendasikan penggunaan nama skema, tabel, kolom, dan kolom kunci utama yang diawali dengan huruf atau underscore serta hanya terdiri dari huruf, angka, dan underscore. Sink menerapkan persyaratan ini saat membuat tabel Lakebase secara otomatis. Untuk menggunakan pengenal yang tidak memenuhi persyaratan ini, buat tabel target sebelum memulai kueri.

Sambungkan ke database

Sink Lakebase mendukung metode koneksi berikut:

Tabel Lakebase terdaftar di Unity Catalog

Untuk tabel Lakebase yang terdaftar di Unity Catalog, konektor secara otomatis mengelola kredensial dan menggunakan identitas pengguna atau perwakilan layanan yang menjalankan kueri. Jika tabel tidak ada, konektor akan membuat tabel.

Untuk mendaftarkan database Lakebase dengan Unity Catalog, lihat Mendaftarkan database Lakebase di Unity Catalog.

Untuk menulis ke tabel Lakebase, gunakan metode .toTable() dengan nama tabel yang sepenuhnya memenuhi syarat, 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>")

Ganti placeholder berikut:

  • <catalog>.<schema>.<table>: Nama lengkap yang memenuhi syarat dari tabel target. catalog adalah katalog Unity Catalog yang Anda buat saat mendaftarkan database Lakebase, lihat Mendaftarkan database Lakebase di Katalog Unity. Jika tabel tidak ada, konektor akan membuatnya.
  • <primary-key-columns>: Opsional. Daftar yang dipisahkan oleh koma dari semua kolom dalam kunci utama tabel target, misalnya id atau user_id,event_type. Lihat perilaku Upsert.
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Jalur volume Unity Catalog tempat kueri menyimpan checkpoint-nya. Anda juga dapat menggunakan URI penyimpanan objek cloud. Lokasi harus berupa penyimpanan yang dapat Anda tulis, bukan disk lokal, dan harus unik untuk setiap kueri streaming. Ini tidak bergantung pada tabel target. Lihat checkpoint Streaming Terstruktur.

Untuk konfigurasi opsional, seperti batchsize dan batchinterval, lihat opsi sink PostgreSQL.

Tabel Lakebase tidak terdaftar di Unity Catalog

Untuk tabel Lakebase yang tidak terdaftar di Unity Catalog, konektor secara otomatis mengelola kredensial dan menggunakan identitas pengguna atau perwakilan layanan yang menjalankan kueri. Jika tabel tidak ada, konektor akan membuat tabel.

Untuk menulis ke tabel Lakebase, gunakan opsi endpoint dan 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()

Ganti placeholder berikut:

  • <project-id>.<branch-id>.<endpoint-id>: Titik akhir Lakebase Anda. Temukan ketiga nilai dalam Nama sumber daya pada menu Dapatkan ID dari tab Komputasi , yang memiliki format projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>. Lihat Pengidentifikasi komputasi.
  • <database>: Opsional. Nama database PostgreSQL target. Secara default menjadi databricks_postgres. Lihat Mengelola database.
  • <schema>.<table>: Tabel target dalam format schema.table. Jika Anda menghilangkan skema, sink menggunakan skema public. Untuk pembuatan tabel otomatis, gunakan pengenal yang dimulai dengan huruf atau garis bawah dan hanya berisi huruf, angka, dan garis bawah.
  • <primary-key-columns>: Opsional. Daftar yang dipisahkan oleh koma dari semua kolom dalam kunci utama tabel target, misalnya id atau user_id,event_type. Lihat perilaku Upsert.
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Jalur volume Unity Catalog tempat kueri menyimpan checkpoint-nya. Anda juga dapat menggunakan URI penyimpanan objek cloud. Lokasi harus berupa penyimpanan yang dapat Anda tulis, bukan disk lokal, dan harus unik untuk setiap kueri streaming. Ini tidak bergantung pada tabel target. Lihat checkpoint Streaming Terstruktur.

Untuk konfigurasi opsional, seperti batchsize dan batchinterval, lihat opsi sink PostgreSQL.

PostgreSQL eksternal dengan kredensial Unity Catalog

Penting

Fitur ini ada di Pratinjau Umum. Admin ruang kerja dapat mengontrol akses ke Custom JDBC pada UC Compute dari halaman Pratinjau. Lihat Kelola Pratinjau Azure Databricks.

Gunakan koneksi Unity Catalog untuk mengautentikasi ke database PostgreSQL eksternal tanpa menyimpan kredensial di kode Anda. Tabel target harus sudah ada.

Buat koneksi bertipe POSTGRESQL, lihat Buat koneksi. Pengguna atau perwakilan layanan yang menjalankan kueri harus memiliki USE CONNECTION pada koneksi tersebut.

Untuk menulis ke tabel PostgreSQL, gunakan opsi databricks.connection, database, dan 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()

Ganti placeholder berikut:

  • <connection-name>: Nama koneksi Unity Catalog.
  • <database>: Nama dari database PostgreSQL tujuan.
  • <schema>.<table>: Tabel target yang sudah ada dalam schema.table format. Jika Anda menghilangkan skema, sink menggunakan skema public.
  • <primary-key-columns>: Opsional. Daftar yang dipisahkan oleh koma dari semua kolom dalam kunci utama tabel target, misalnya id atau user_id,event_type. Lihat perilaku Upsert.
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Jalur volume Unity Catalog tempat kueri menyimpan checkpoint-nya. Anda juga dapat menggunakan URI penyimpanan objek cloud. Lokasi harus berupa penyimpanan yang dapat Anda tulis, bukan disk lokal, dan harus unik untuk setiap kueri streaming. Ini tidak bergantung pada tabel target. Lihat checkpoint Streaming Terstruktur.

Koneksi PostgreSQL selalu menggunakan TLS. Verifikasi sertifikat mengikuti pengaturan pada koneksi Unity Catalog, yang Anda pilih saat membuat koneksi:

  • Sertifikat Trust server: Saat dipilih, koneksi menggunakan sslmode=require, yang mengenkripsi koneksi tanpa memverifikasi sertifikat server.
  • Sertifikat server yang disediakan pengguna: Berikan sertifikat server yang dikodekan PEM untuk digunakan sslmode=verify-full saat sertifikat Trust server tidak dipilih. Jika Anda tidak menyediakan sertifikat, koneksi menggunakan sslmode=verify-full dengan trust store bawaan JVM.

Opsi konfigurasi

Sink menghasilkan error jika ada opsi yang tidak dikenali, JDBC_STREAMING_SINK_INVALID_OPTIONS.

Untuk opsi konfigurasi sink, termasuk opsi umum dan opsi untuk setiap metode koneksi, lihat opsi sink PostgreSQL.

Pemetaan jenis data

Sink memeriksa apakah setiap kolom DataFrame kompatibel dengan kolom target padanannya sebelum menulis data ke tabel Lakebase atau PostgreSQL eksternal yang sudah ada.

Tabel berikut berisi tipe yang didukung di Databricks Runtime 18 LTS dan ke atas:

Tipe Spark Tipe tabel Lakebase yang dibuat secara otomatis Tipe kompatibel dalam tabel PostgreSQL yang sudah ada
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, VariantType, NullType jsonb json, jsonb

Tabel berikut berisi tipe yang didukung di Databricks Runtime 19 dan ke atas:

Tipe Spark Tipe tabel Lakebase yang dibuat secara otomatis Tipe kompatibel dalam tabel PostgreSQL yang sudah ada
DayTimeIntervalType, YearMonthIntervalType interval interval

Perilaku Upsert

Opsi ini upsertkey mengidentifikasi kolom kunci utama tabel target. Untuk tabel yang sudah ada, kolom-kolom dalam upsertkey harus persis sesuai dengan kunci utama tabel tersebut. Jika Anda menghilangkan opsi tersebut, sink akan membaca kunci utama dari tabel. Untuk tabel Lakebase yang dibuat oleh sink, upsertkey mendefinisikan kunci utama. Jika Anda menghilangkan opsi tersebut, sink akan membuat tabel tanpa kunci primer.

Ketika tabel target memiliki kunci utama, sink melakukan upsert dengan sintaks INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ... PostgreSQL. Ketika tabel target tidak memiliki kunci utama, sink melakukan penyisipan. Mode keluaran kueri tidak memengaruhi perilaku ini.

Semua kolom primary key harus ada di DataFrame dan menggunakan tipe yang sebanding, seperti tipe numerik atau string.

Untuk mengontrol bagaimana upsert memperbarui kolom non-primary key, lihat Merge strategies for upserts.

Strategi merge untuk upsert

Di Databricks Runtime 19 dan ke atas, Anda dapat menggunakan strategi merge untuk mengontrol bagaimana sink memperbarui kolom non-primary key ketika baris memiliki primary key yang sama. Misalnya, Anda dapat menyimpan timestamp terbesar atau memperbarui kolom hanya ketika event yang masuk memiliki nilai pengurutan yang lebih baru.

Strategi penggabungan hanya berlaku untuk upsert. Jika tabel target tidak memiliki kunci utama, pengaturan upsert.mergeStrategies atau upsert.defaultMergeStrategy menyebabkan kesalahan. Anda hanya dapat menetapkan strategi untuk kolom yang bukan primary key. Atur upsert.mergeStrategies ke daftar yang dipisahkan koma dalam format column:strategy . Atur upsert.defaultMergeStrategy untuk mengonfigurasi default untuk kolom tanpa strategi eksplisit. upsert.defaultMergeStrategy tidak menerima latestBy atau earliestBy. Secara default, sink menggunakan strategi latest.

Strategi penggabungan yang didukung

Strategy Behavior
latest Gunakan nilai terbaru berdasarkan urutan kedatangan. Ini adalah default.
earliest Gunakan nilai paling awal berdasarkan urutan kedatangan.
latestNonNull Gunakan nilai non-null terbaru berdasarkan urutan kedatangan.
earliestNonNull Gunakan nilai non-null paling awal berdasarkan urutan kedatangan.
greatest Gunakan nilai yang lebih besar di antara nilai yang ada dan nilai yang masuk. Nilai null diabaikan, sehingga nilai masuk null tidak menggantikan nilai yang sudah ada.
least Gunakan nilai yang lebih rendah antara nilai yang ada dan nilai yang masuk. Nilai null diabaikan, sehingga nilai masuk null tidak menggantikan nilai yang sudah ada.
latestBy(column) Gunakan nilai dari baris dengan nilai terbesar di kolom pengurutan yang ditentukan. Lihat Bagaimana latestBy dan earliestBy menangani seri serta null.
earliestBy(column) Gunakan nilai dari baris dengan nilai terendah di kolom pengurutan yang ditentukan. Lihat Bagaimana latestBy dan earliestBy menangani seri serta null.

Bagaimana latestBy dan earliestBy menangani tie dan null

latestBy(column) dan earliestBy(column) pilih nilai dengan membandingkan nilai pengurutan yang masuk dengan yang sudah tersimpan di tabel. Untuk menjaga nilai yang tersimpan tetap akurat, kolom pengurutan harus menggunakan strategi pencocokan, baik di dalam upsert.mergeStrategies maupun melalui upsert.defaultMergeStrategy:

  • Untuk latestBy(column), kolom pengurutan harus menggunakan greatest.
  • Untuk earliestBy(column), kolom pengurutan harus menggunakan least.

Misalnya, is_success:latestBy(last_updated_at) memerlukan last_updated_at:greatest.

Ketika kedua nilai pengurutan tidak bernilai null, latestBy(column) memperbarui kolom hanya ketika nilai pengurutan yang masuk secara mutlak lebih besar (>) daripada nilai pengurutan yang ada. Untuk earliestBy(column), harus lebih kecil secara ketat (<). Untuk kedua strategi, ketika nilai urutan sama, sink mempertahankan nilai kolom yang ada. Dalam satu batch, jika beberapa baris memiliki kunci utama dan nilai urutan yang sama, sink mempertahankan baris pertama yang diterimanya.

Misalnya, jika is_success menggunakan latestBy(last_updated_at) dan baris masuk memiliki last_updated_at yang sama dengan baris yang ada, sink mempertahankan is_success yang ada, meskipun nilai masuk berbeda.

Untuk kedua strategi, nilai pengurutan non-null lebih diutamakan daripada nilai urutan null. Baris masuk dengan nilai pengurutan non-null menggantikan nilai yang sudah ada dengan nilai urutannya null. Baris masuk dengan nilai pengurutan null tidak menggantikan nilai yang sudah ada, termasuk ketika nilai pengurutan yang ada juga null.

Contoh: Model logika upsert kustom

Pertimbangkan sebuah tabel dengan kolom-kolom id, last_updated_at, creation_date, dan is_success. Anda mungkin ingin mengekspresikan logika berikut:

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;

Untuk memodelkan logika ini, gunakan strategi penggabungan berikut:

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

Pengoptimalan kinerja

Batching dan backpressure

Flush dipicu ketika salah satu kondisi terpenuhi:

  • Buffer mencapai batchsize baris, yang nilai defaultnya adalah 1000.
  • Umur buffer melebihi batchinterval, yang nilai defaultnya adalah 100 milliseconds.

Ketika basis data tidak dapat mengimbangi laju data yang masuk, sink meneruskan backpressure ke hulu menuju sumber.

Panduan latensi dan throughput:

  • Untuk beban kerja latensi rendah dengan mode real time, kurangi batchinterval untuk menjamin waktu maksimum yang lebih singkat sebelum pembilasan. Lihat konsep mode waktu nyata untuk konsep dan contoh mode waktu nyata untuk contoh kode.
  • Untuk beban kerja dengan throughput tinggi, tingkatkan batchsize guna mengurangi overhead pada setiap transaksi.

Perilaku koneksi

Sink menggunakan pengumpulan koneksi pada pelaksana. Secara default, setiap tugas menggunakan satu koneksi database.

Databricks merekomendasikan agar Anda menggunakan nilai default 1 task untuk setiap koneksi. Jika Anda meningkatkan jumlah tugas untuk setiap koneksi, Anda dapat menyebabkan ketidakcocokan koneksi dan meningkatkan latensi untuk koneksi throughput tinggi.

Untuk mengonfigurasi rasio tugas terhadap koneksi, atur konfigurasi Spark spark.databricks.sql.streaming.jdbc.tasksPerConnection. Jika database target memiliki batas koneksi rendah, kurangi jumlah partisi acak atau tingkatkan spark.databricks.sql.streaming.jdbc.tasksPerConnection.

Sink secara otomatis mencoba kembali kesalahan JDBC sementara, termasuk kegagalan koneksi, kebuntuan, dan pembatasan laju. Jika sink menghabiskan semua upaya percobaan ulang, kueri akan gagal.

Pemicu dan mode output yang didukung

Triggers

Tabel ini menunjukkan dukungan untuk tipe pemicu Structured Streaming pada komputasi klasik dan serverless:

Pemicu Komputasi tradisional Komputasi tanpa server (notebook dan pekerjaan)
RealTime Yes No
ProcessingTime Yes No
AvailableNow Yes Yes
Once Yes. Deprecated. Gunakan AvailableNow. Yes. Deprecated. Gunakan AvailableNow.

Mode keluaran

Tabel ini memperlihatkan dukungan untuk mode output Streaming Terstruktur:

Mode keluaran Dukungan
update Yes
append Yes. Perilaku identik dengan update. Kueri melakukan upsert saat tabel target memiliki kunci primer; jika tidak, kueri melakukan penyisipan. Lihat perilaku Upsert.
complete No

Batasan

  • Untuk database PostgreSQL eksternal yang terhubung melalui koneksi Unity Catalog, tabel target harus sudah ada. Sink secara otomatis membuat tabel yang belum ada hanya di Lakebase.
  • Pipeline Lakeflow tidak didukung.
  • Terakhir diperbarui pada