Membuat konektor Sumber Cloud SQL untuk PostgreSQL

Konektor Sumber Cloud SQL untuk PostgreSQL memungkinkan Anda menangkap perubahan tingkat baris dari database Cloud SQL untuk PostgreSQL dan menuliskannya ke topik di cluster Managed Service untuk Apache Kafka.

Kasus penggunaan umum untuk konektor ini meliputi:

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

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 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)
  • Cloud SQL Instance User (roles/cloudsql.instanceUser)

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, lihat Membuat cluster Connect di project yang berbeda.

Mengonfigurasi database

Sebelum membuat konektor, Anda harus mengonfigurasi replikasi database dan mengaktifkan konektor untuk melakukan autentikasi dengan database. Bagian berikut 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. Dalam daftar Choose a flag, pilih cloudsql.logical_decoding.

  7. Untuk Value, pilih On.

  8. Klik Done.

  9. Klik Simpan.

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

Mengonfigurasi pengambilan data perubahan (CDC)

Setelah mengaktifkan decoding logis di instance, aktifkan change data capture (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 yang akan direplikasi.

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

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

  • Opsi 2. Membuat publikasi untuk kumpulan 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 nama skema dan tabel dalam tanda kutip ganda, seperti yang ditunjukkan, akan 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.

Membuat 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 Users.

  4. Klik Add user account.

  5. Di panel Add a user account, pilih Cloud IAM.

  6. Di kolom IAM principal, masukkan berikut ini:

    service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com

    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 dari cluster Connect.

  • INSTANCE_NAME: Nama instance Cloud SQL untuk PostgreSQL.

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

Mengonfigurasi akun pengguna

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

Konsol

  1. Memungkinkan pengguna membaca write-ahead log.

    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 tabel satu per satu. Jika memilih opsi ini, Anda juga harus menetapkan 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 Konfigurasikan 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 Mengupdate cluster Connect.
IP Publik
  1. Konfigurasikan IP publik untuk instance Anda. Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi IP publik.
  2. Siapkan Public NAT untuk memungkinkan worker cluster Connect berkomunikasi dengan internet. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan Public NAT. Saat Anda 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, mengirimkan 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 panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka.

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

  6. Dalam daftar Instance, pilih instance Cloud SQL.

  7. Dalam 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 Table names, masukkan daftar tabel yang dipisahkan koma untuk membaca data perubahan, 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 setelan 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 Task restart policy. 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 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 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 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 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 pada 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 yang diinginkan; 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 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 menstreaming 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 ditetapkan secara default ke debezium, kecuali jika dikonfigurasi secara berbeda menggunakan properti slot.name.

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

Filter tabel

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

Apache Kafka® adalah merek dagang terdaftar dari Apache Software Foundation atau afiliasinya di Amerika Serikat dan/atau negara lainnya.