Membuat konektor Sumber Cloud SQL untuk PostgreSQL

Dokumen ini menjelaskan cara membuat konektor Sumber Cloud SQL untuk PostgreSQL untuk Kafka Connect.

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

Kasus penggunaan untuk konektor ini meliputi:

  • Pantau 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.

Sebelum memulai

Sebelum membuat konektor Sumber Cloud SQL untuk PostgreSQL, pastikan Anda memiliki hal berikut:

Peran dan izin yang diperlukan

Untuk mendapatkan izin yang diperlukan guna 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:

  • Buat konektor: managedkafka.connectors.create

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

Memberikan izin untuk membaca dari Cloud SQL

Akun layanan Managed Kafka harus memiliki izin untuk mengakses Cloud SQL untuk PostgreSQL. Berikan peran IAM berikut ke akun layanan:

  • Cloud SQL Client (roles/cloudsql.client)
  • Pengguna Instance Cloud SQL (roles/cloudsql.instanceUser)

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

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

Konfigurasi database

Sebelum membuat konektor, Anda harus mengonfigurasi replikasi database dan mengaktifkan konektor untuk mengautentikasi dengan database. Bagian berikut akan menjelaskan langkah-langkah ini.

Mengaktifkan decoding logis

Konektor Sumber Cloud SQL untuk PostgreSQL mengandalkan fitur dekode logis PostgreSQL. Untuk mengaktifkan decoding logis di instance Cloud SQL untuk PostgreSQL, lakukan langkah-langkah berikut.

Konsol

  1. Buka Cloud SQL > Instances.

    Buka Instances

  2. Klik nama instance.

  3. Klik Edit.

  4. Luaskan Flags and parameters.

  5. Klik Tambahkan flag database.

  6. Di daftar Choose a flag, pilih cloudsql.logical_decoding.

  7. Untuk Nilai, pilih On.

  8. Klik Done.

  9. Klik Simpan.

Untuk mengetahui informasi selengkapnya, lihat Menyiapkan replikasi dan decoding logis.

Mengonfigurasi pengambilan data perubahan (CDC)

Setelah Anda mengaktifkan decoding logis di instance, aktifkan pengambilan data perubahan (CDC) untuk tabel yang ingin Anda replikasi.

Untuk mengaktifkan CDC untuk tabel, jalankan pernyataan SQL CREATE PUBLICATION. Pernyataan ini membuat publikasi, yang menentukan grup tabel untuk direplikasi.

  • Opsi 1. Buat publikasi yang mereplikasi perubahan untuk semua tabel dalam database.

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

  • Opsi 2. Buat publikasi untuk sekumpulan tabel tertentu.

    CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;

    Ganti TABLE_LIST dengan daftar tabel yang dipisahkan koma, dalam format "schema_name"."table_name". Mengapit skema dan nama tabel dalam tanda kutip ganda, seperti yang ditunjukkan, mencegah error sintaksis jika nama berisi karakter khusus atau huruf besar.

Secara default, konektor menggunakan dbz_publication untuk nama publikasi. Untuk menggunakan publikasi dengan nama yang berbeda, lihat Nama publikasi.

Buat akun pengguna untuk akun layanan Managed Kafka

Konektor Sumber Cloud SQL untuk PostgreSQL menggunakan autentikasi database IAM untuk terhubung ke database. Untuk mengaktifkan autentikasi database IAM, tambahkan akun layanan Managed Kafka ke instance Cloud SQL, sebagai berikut:

Konsol

  1. Buka Cloud SQL > Instances

    Buka Instances

  2. Klik nama instance.

  3. Di panel navigasi, klik Pengguna.

  4. Klik Add user account.

  5. Di panel Tambahkan akun pengguna, pilih Cloud IAM.

  6. Di kolom IAM principal, masukkan hal berikut:

    service-PROJECT_NUMBER@gcp-sa-managedkafka.

    Ganti PROJECT_NUMBER dengan nomor project cluster Connect.

  7. Klik Tambahkan.

gcloud

Jalankan perintah gcloud sql users create:

gcloud sql users create service-PROJECT_NUMBER@gcp-sa-managedkafka.iam \
  --instance=INSTANCE_NAME \
  --type=cloud_iam_service_account

Ganti kode berikut:

  • PROJECT_NUMBER: Nomor project cluster Connect.

  • INSTANCE_NAME: Nama instance Cloud SQL untuk PostgreSQL.

Karena batas panjang pada nama pengguna database, akhiran . dihapus dari nama pengguna, sehingga nama pengguna menjadi service-PROJECT_NUMBER@gcp-sa-managedkafka.iam. Saat Anda menjalankan kueri SQL yang mereferensikan akun pengguna IAM, tentukan nama yang dipangkas.

Mengonfigurasi akun pengguna

Setelah membuat akun pengguna IAM, hubungkan ke database sebagai pengguna dengan peran cloudsqlsuperuser (seperti pengguna postgres default), lalu jalankan kueri SQL berikut.

Konsol

  1. Memungkinkan pengguna membaca log tulis-dahulu.

    ALTER USER "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam" WITH REPLICATION;

  2. Berikan izin SELECT kepada pengguna di tabel.

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME"
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

    Atau, Anda dapat memberikan izin SELECT pada setiap tabel. Jika Anda memilih opsi ini, Anda juga harus menyetel properti konfigurasi table.include.list konektor ke daftar tabel yang diizinkan. Kueri SQL berikut memberikan izin SELECT pada satu tabel:

    GRANT SELECT ON TABLE SCHEMA_NAME.TABLE_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

  3. Untuk setiap tabel, beri pengguna akses ke skema tabel. Anda dapat melewati langkah ini jika tabel berada dalam skema public default.

    GRANT USAGE ON SCHEMA SCHEMA_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

Mengonfigurasi jaringan

Konektor Sumber Cloud SQL untuk PostgreSQL dapat terhubung ke instance Cloud SQL dengan cara berikut:

  • IP Pribadi
  • Private Service Connect
  • IP Publik

Untuk mengetahui informasi selengkapnya tentang opsi ini, lihat Memilih cara terhubung ke Cloud SQL. Sebagai praktik terbaik keamanan, sebaiknya gunakan IP pribadi atau Private Service Connect, karena opsi ini tidak memerlukan koneksi ke alamat IP eksternal.

Tabel berikut menunjukkan persyaratan jaringan untuk setiap opsi:

Jenis alamat IP Persyaratan
IP Pribadi Konfigurasi IP pribadi untuk instance Anda. Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi IP pribadi.
Private Service Connect
  1. Konfigurasi Private Service Connect untuk instance Anda dan dapatkan nama DNS endpoint Private Service Connect. Untuk mengetahui informasi selengkapnya, lihat Menghubungkan ke instance menggunakan Private Service Connect.
  2. Tambahkan nama DNS endpoint ke Domain DNS yang dapat di-resolve cluster Connect. Untuk mengetahui informasi selengkapnya, lihat Memperbarui cluster Connect.
IP Publik
  1. Konfigurasi IP publik untuk instance Anda. Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi IP publik.
  2. Siapkan NAT Publik untuk memungkinkan pekerja cluster Connect berkomunikasi dengan internet. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan Public NAT. Saat membuat gateway Cloud NAT, tentukan Jaringan VPC yang berisi subnet utama cluster Connect.

Membuat konektor Sumber Cloud SQL untuk PostgreSQL

Untuk membuat konektor Sumber Cloud SQL untuk PostgreSQL, 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 konektor berjalan, konektor akan terus mengirim peristiwa perubahan ke topik. Untuk mengetahui informasi selengkapnya tentang snapshot awal, lihat Snapshot dalam dokumentasi Debezium.

Konsol

  1. Di konsol Google Cloud , 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 for Apache Kafka.

  5. Untuk Connector plugin, pilih Cloud SQL untuk PostgreSQL Source.

  6. Di daftar Instance, pilih instance Cloud SQL.

  7. Di daftar Database, pilih database Cloud SQL.

  8. Di kolom Topic prefix, masukkan awalan yang akan digunakan untuk nama topik Kafka. Pilih awalan unik untuk setiap konektor Sumber Cloud SQL untuk PostgreSQL.

  9. Opsional: Di kolom Nama tabel, masukkan daftar tabel yang dipisahkan koma untuk dibaca perubahan datanya, dalam format "schema_name"."table_name". Jika Anda mengosongkan kolom ini, konektor akan membaca data perubahan dari semua tabel non-sistem dalam database.

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

    Anda mungkin perlu mengganti nilai default untuk properti berikut:

    • driver.ipTypes: Properti ini harus cocok dengan konfigurasi jaringan instance Cloud SQL Anda. Lihat Jenis alamat IP.

    • slot.name: Jika Anda membuat beberapa instance konektor untuk database yang sama, tentukan nilai unik untuk setiap konektor. Lihat Slot replikasi.

  11. Opsional: Pilih Kebijakan mulai ulang tugas. Untuk mengetahui informasi selengkapnya, lihat Kebijakan memulai ulang tugas.

  12. Klik Create.

gcloud

  1. Di konsol Google Cloud , aktifkan Cloud Shell.

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan perintah 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 pada 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 for Apache Kafka. Nama konektor tidak dapat diubah.

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

    • 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 Cloud SQL untuk PostgreSQL:

    connector.class: io.debezium.connector.postgresql.PostgresConnector
database.dbname: DATABASE_NAME
driver.cloudSqlInstance: INSTANCE_ID
driver.enableIamAuth: "true"
driver.ipTypes: IP_TYPES
driver.sslmode: disable
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: "false"
plugin.name: pgoutput
slot.name: SLOT_NAME
table.include.list: TABLE_LIST
topic.prefix: TOPIC_PREFIX
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: "true"

    Ganti kode berikut:

    • INSTANCE_ID: ID instance Cloud SQL yang berisi database, dengan format sebagai berikut:

      PROJECT_ID:REGION:INSTANCE_NAME

    • DATABASE_NAME: Nama database Cloud SQL yang akan dibaca.

    • IP_TYPES: Daftar jenis alamat IP yang dipisahkan koma

    • SLOT_NAME: Nama slot replikasi yang akan dibuat.

    • TABLE_LIST: Daftar tabel yang dipisahkan koma untuk membaca data perubahan, dalam format "schema_name"."table_name".

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

Mengonfigurasi konektor

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

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.

Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi jaringan.

Nama publikasi

Secara default, konektor mencoba melakukan streaming dari publikasi bernama dbz_publication. Untuk menentukan publikasi lain, 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 menghapus 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 hapus slot replikasi yang tidak digunakan.

Filter tabel

Secara default, konektor mengambil data perubahan dari setiap tabel non-sistem dalam database. Untuk memfilter tabel yang direkam, 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