Membuat konektor Sumber PostgreSQL Generik

Dokumen ini menjelaskan cara membuat konektor Sumber PostgreSQL Generik.

Konektor Sumber PostgreSQL Generik adalah instance konektor PostgreSQL Debezium. Konektor ini membaca perubahan tingkat baris dari database PostgreSQL dan menuliskannya ke topik di cluster Managed Service untuk Apache Kafka.

Kasus penggunaan untuk konektor ini mencakup:

  • Memantau perubahan database tingkat baris secara real time.
  • Mengintegrasikan peristiwa perubahan database ke dalam arsitektur berbasis peristiwa.
  • Merespons peristiwa database seperti penyisipan atau penghapusan baris.
  • Menyalin perubahan database ke sistem lain.
  • Mereplikasi atau memulihkan tabel PostgreSQL.

Sebelum memulai

Sebelum membuat konektor Sumber PostgreSQL Generik, pastikan Anda memiliki hal berikut:

  • Database PostgreSQL.

  • Cluster Connect yang terkait dengan cluster Kafka.

  • Buat secret Secret Manager yang menyimpan sandi database. Jika konfigurasi Anda menggunakan SSL database, buat juga secret untuk sandi SSL database. Konfigurasi cluster Connect Anda dengan secret. Untuk mengetahui informasi selengkapnya, lihat resource Secret Manager.

Peran dan izin yang diperlukan

Untuk mendapatkan izin yang diperlukan untuk membuat konektor, minta administrator Anda untuk memberi Anda peran IAM Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) di project Anda. Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses ke project, folder, dan organisasi.

Peran bawaan ini berisi izin yang diperlukan untuk membuat konektor. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk membuat konektor:

  • Membuat konektor: managedkafka.connectors.create

Anda mungkin juga bisa mendapatkan izin ini dengan peran khusus atau peran bawaan lainnya.

Memberikan izin untuk mengakses secret Secret Manager

Akun layanan Managed Kafka memerlukan izin untuk melihat dan mengakses secret yang disimpan di Secret Manager. Berikan peran IAM berikut ke akun layanan:

  • Secret Manager Viewer (roles/secretmanager.viewer)
  • Secret Manager Secret Accessor (roles/secretmanager.secretAccessor)

Akun layanan Managed Kafka memiliki format berikut: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, dengan PROJECT_NUMBER adalah nomor project cluster Connect.

Jika cluster Connect Anda berada di project yang berbeda dengan cluster Managed Service untuk Apache Kafka cluster, lihat Membuat cluster Connect di project yang berbeda.

Mengonfigurasi database PostgreSQL

Untuk mengaktifkan konektor agar dapat membaca peristiwa perubahan data dari database Anda, konfigurasikan setelan berikut.

  1. Tetapkan wal_level server ke logical.

    ALTER SYSTEM SET wal_level = logical;

    Mulai ulang server untuk menerapkan setelan.

  2. Buat pengguna database untuk konektor agar dapat melakukan autentikasi ke PostgreSQL. Pengguna database harus memiliki peran replikasi, sehingga dapat terhubung ke server dalam mode replikasi.

    CREATE ROLE ROLE_NAME WITH REPLICATION LOGIN PASSWORD 'ROLE_PASSWORD';

    Ganti kode berikut:

    • ROLE_NAME: Nama pengguna, misalnya debezium_user.
    • ROLE_PASSWORD: Sandi pengguna.

  3. Buat publikasi untuk tabel yang ingin Anda ambil. Konektor berlangganan publikasi untuk menerima peristiwa perubahan data.

    CREATE PUBLICATION dbz_publication FOR TABLE "SCHEMA_NAME"."TABLE_NAME";

    Ganti kode berikut:

    • SCHEMA_NAME: Skema tabel.

    • TABLE_NAME: Nama tabel.

    Sebaiknya sertakan nama skema dan tabel dalam tanda kutip ganda, seperti yang ditunjukkan, untuk menghindari error sintaksis jika nama berisi karakter khusus atau huruf besar.

    Atau, Anda dapat membuat publikasi yang mereplikasi perubahan untuk semua tabel dalam database:

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

    Bergantung pada setelan publication.autocreate.mode konektor, Anda dapat membuat publikasi secara manual atau membiarkan konektor membuatnya secara otomatis. Untuk mengetahui informasi selengkapnya, lihat Mode publikasi.

  4. Untuk setiap tabel, berikan hak istimewa SELECT pada tabel kepada pengguna database.

    GRANT SELECT ON TABLE "SCHEMA_NAME"."TABLE_NAME" TO ROLE_NAME;

    Atau, Anda dapat memberikan hak istimewa pilih pada semua tabel dalam skema:

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

  5. Untuk setiap tabel, berikan hak istimewa USAGE pada skema tabel kepada pengguna database. Anda dapat melewati langkah ini jika tabel berada dalam skema public default.

    GRANT USAGE ON SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

Membuat konektor Sumber PostgreSQL Generik

Untuk membuat konektor Sumber PostgreSQL Generik, lakukan langkah-langkah berikut.

Saat diinisialisasi, konektor akan melakukan tindakan berikut:

  1. Membuat snapshot awal database.
  2. Membuat topik Kafka untuk setiap tabel yang memiliki baris.
  3. Untuk setiap baris database, mengirim peristiwa perubahan ke topik yang sesuai.

Saat berjalan, konektor akan terus mengirim peristiwa perubahan ke topik. Untuk mengetahui informasi selengkapnya tentang snapshot awal, lihat Snapshot dalam dokumentasi Debezium.

Konsol

  1. Di Google Cloud konsol, buka halaman Connect Clusters.

    Buka Connect Clusters

  2. Klik cluster Connect tempat Anda ingin membuat konektor.

  3. Klik Create connector.

  4. Untuk nama konektor, masukkan string.

    Untuk mengetahui panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka.

  5. Untuk Connector plugin, pilih Generic PostgreSQL Source.

  6. Di kolom Database hostname, masukkan nama host atau alamat IP server PostgreSQL.

  7. Di kolom Database name, masukkan nama database.

  8. Di kolom Database user, masukkan nama peran replika. Konektor melakukan autentikasi ke server PostgreSQL menggunakan peran ini.

  9. Di kolom Topic prefix, masukkan awalan yang akan digunakan untuk nama topik Kafka.

  10. Di daftar Secret, pilih secret yang berisi sandi database.

  11. Opsional: Di kotak Configurations, tambahkan properti konfigurasi atau edit properti default. Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi konektor.

  12. Opsional: Pilih Task restart policy. Untuk mengetahui informasi selengkapnya, lihat Kebijakan memulai ulang tugas.

  13. Klik Create.

gcloud

  1. Di Google Cloud konsol, aktifkan Cloud Shell.

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi sesi.

  2. Jalankan perintah gcloud managed-kafka connectors create:

    gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE

    Ganti kode berikut:

    • CONNECTOR_ID: ID atau nama konektor. Untuk mengetahui panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka. Nama konektor tidak dapat diubah.

    • LOCATION: Lokasi tempat Anda membuat konektor. Lokasi ini harus sama dengan lokasi tempat Anda membuat cluster Connect.

    • CONNECT_CLUSTER_ID: ID cluster Connect tempat konektor dibuat.

    • CONFIG_FILE: Jalur ke file konfigurasi YAML untuk konektor.

    Berikut adalah contoh file konfigurasi untuk konektor Sumber PostgreSQL Generik:

    connector.class: io.debezium.connector.postgresql.PostgresConnector
database.dbname: DATABASE_NAME
database.hostname: HOSTNAME
database.password: CREDENTIALS
database.user: DATABASE_USER
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: "false"
plugin.name: pgoutput
topic.prefix: TOPIC_PREFIX
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: "true"

    Ganti kode berikut:

    • HOSTNAME: Nama host database PostgreSQL yang akan dibaca.

    • DATABASE_NAME: Nama database PostgreSQL yang akan dibaca.

    • DATABASE_USER: Pengguna database PostgreSQL yang akan digunakan saat melakukan autentikasi ke database.

    • CREDENTIALS: Jalur ke secret Secret Manager yang berisi sandi database. Tentukan secret menggunakan format berikut:

      ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}

    • TOPIC_PREFIX: Awalan yang akan digunakan untuk nama topik Kafka.

Mengonfigurasi konektor

Bagian ini menjelaskan beberapa properti konfigurasi yang dapat Anda tetapkan pada konektor. Untuk mengetahui daftar lengkapnya, lihat Konektor Debezium untuk PostgreSQL dalam dokumentasi Debezium.

Konfigurasi sandi dan sandi SSL

Hanya jalur secret yang didukung dalam konfigurasi database.password dan database.sslpassword. Backend mengharapkan konfigurasi ini menggunakan format berikut: ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}.

Jenis alamat IP

Properti driver.ipTypes menentukan jenis alamat IP yang digunakan konektor untuk terhubung ke database:

  • PRIVATE: IP Pribadi
  • PSC: Private Service Connect
  • PUBLIC: IP Publik

Properti driver.ipTypes berisi daftar jenis IP yang dipisahkan koma dalam urutan pilihan ; misalnya, driver.ipTypes=PRIVATE,PUBLIC.

Mode publikasi

Konektor Sumber PostgreSQL Generik melakukan streaming peristiwa perubahan dari a publikasi di database. Anda dapat membuat publikasi secara manual, atau membiarkan konektor membuatnya secara otomatis.

Setelan publication.autocreate.mode menentukan cara dan apakah konektor harus membuat publikasi.

  • filtered. Jika publikasi tidak ada, konektor akan membuat publikasi baru yang hanya menyertakan tabel yang diambil. Pengguna database harus memiliki izin CREATE di database dan menjadi pemilik tabel yang disertakan.

    Jika publikasi sudah ada, konektor akan mengubahnya untuk menyertakan tabel yang diambil. Untuk mengubah publikasi yang ada, pengguna database harus menjadi pemilik publikasi dan pemilik tabel yang disertakan.

  • all_tables. Jika publikasi tidak ada, konektor akan membuat publikasi baru menggunakan parameter FOR ALL TABLES. Pengguna database harus menjadi pengguna super.

    Peran pengguna super melewati semua pemeriksaan izin dalam database, sehingga tidak direkomendasikan untuk memberikan SUPERUSER kepada pengguna database. Sebagai gantinya, buat publikasi secara manual atau tetapkan publication.autocreate.mode=filtered.

  • disabled. Jika publikasi tidak ada, akan terjadi error. Konektor tidak membuat publikasi baru.

Nilai defaultnya adalah all_tables.

Nama publikasi

Secara default, konektor mencoba melakukan streaming dari publikasi bernama dbz_publication. Untuk menentukan publikasi yang berbeda, tambahkan publication.name=PUBLICATION_NAME ke konfigurasi, dengan PUBLICATION_NAME adalah nama publikasi. Contoh: publication.name=my_publication.

Slot replikasi

PostgreSQL menggunakan slot replikasi untuk melakukan streaming perubahan tabel database. Secara default, konektor membuat slot replikasi bernama debezium. Untuk menggunakan nama slot yang berbeda, tetapkan properti slot.name.

Jika Anda membuat dua instance konektor untuk database yang sama, Anda harus menentukan nama slot unik untuk setiap konektor.

Secara default, konektor menetapkan properti slot.drop.on.stop ke false untuk mencegah kehilangan data. Saat Anda menghapus konektor secara permanen, Anda harus menghentikan slot replikasi yang digunakan konektor secara manual. Nama slot replikasi secara default adalah debezium, kecuali jika dikonfigurasi secara berbeda menggunakan properti slot.name.

Sebaiknya siapkan pemberitahuan untuk memantau penggunaan disk WAL di server database PostgreSQL sumber Anda, dan hentikan slot replikasi yang tidak digunakan.

Filter tabel

Secara default, konektor mengambil data perubahan dari setiap tabel non-sistem dalam database. Untuk memfilter tabel mana yang diambil, tentukan satu atau beberapa setelan berikut:

  • schema.include.list. Daftar skema yang akan disertakan.
  • schema.exclude.list. Daftar skema yang akan dikecualikan. Tidak dapat digunakan dengan schema.include.list.
  • table.include.list. Daftar tabel yang akan disertakan.
  • table.exclude.list. Daftar tabel yang akan dikecualikan. Tidak dapat digunakan dengan table.include.list.

Nama topik

Secara default, konektor membuat topik Kafka dengan konvensi penamaan berikut: topic_prefix.schema.table_name, dengan topic.prefix adalah nilai konfigurasi topic.prefix.

Untuk mengetahui informasi selengkapnya, lihat Nama topik dalam dokumentasi Debezium.

Langkah berikutnya