PolarDB:Konektor Debezium untuk PolarDB for PostgreSQL (Kompatibel dengan Oracle)

Karena PolarDB for PostgreSQL (Compatible with Oracle) dan PostgreSQL komunitas hanya berbeda dalam penanganan beberapa tipe data dan objek bawaan, topik ini menjelaskan cara membangun konektor Debezium yang mendukung PolarDB for PostgreSQL (Compatible with Oracle) dengan menyesuaikan konektor Debezium PostgreSQL dari komunitas menggunakan perubahan kode minimal.

Build konektor Debezium PolarDBO

Penggunaan

Konektor Debezium PolarDBO membaca perubahan inkremental dari database PolarDB for PostgreSQL (Compatible with Oracle) menggunakan replikasi logis. Kondisi berikut harus dipenuhi sebelum Anda dapat menggunakan konektor:

• Parameter kluster wal_level harus diatur ke logical. Pengaturan ini menambahkan informasi yang diperlukan untuk replikasi logis ke dalam write-ahead logging (WAL).

  Catatan

  Anda dapat mengatur parameter kluster wal_level di Konsol. Untuk informasi lebih lanjut, lihat Konfigurasi parameter kluster. Mengubah parameter ini akan merestart kluster. Rencanakan operasi ini secara hati-hati sesuai kebutuhan bisnis Anda.

• Jalankan perintah ALTER TABLE schema.table REPLICA IDENTITY FULL; untuk mengatur REPLICA IDENTITY setiap tabel yang dilangganan menjadi FULL. Hal ini memastikan bahwa event insert dan update berisi nilai sebelumnya untuk semua kolom, sehingga menjamin konsistensi data.

  Catatan

  • REPLICA IDENTITY adalah pengaturan tingkat tabel khusus PostgreSQL. Pengaturan ini menentukan apakah plug-in decoding logis menyertakan nilai sebelumnya dari kolom tabel terkait selama event INSERT dan UPDATE. Untuk informasi lebih lanjut mengenai nilai-nilai REPLICA IDENTITY, lihat REPLICA IDENTITY.

  • Mengatur REPLICA IDENTITY tabel yang dilangganan ke FULL mungkin memerlukan penguncian tabel, yang dapat memengaruhi layanan Anda. Rencanakan operasi ini sesuai kebutuhan bisnis Anda. Anda dapat menjalankan perintah berikut untuk memeriksa apakah pengaturan saat ini sudah FULL:

    SELECT relreplident = 'f' FROM pg_class WHERE relname = 'tablename';

• Nilai parameter max_wal_senders dan max_replication_slots harus lebih besar daripada jumlah total slot replikasi yang sedang digunakan ditambah yang dibutuhkan oleh pekerjaan Kafka Anda.

• Gunakan akun istimewa atau akun standar yang memiliki izin LOGIN dan REPLICATION. Akun tersebut juga harus memiliki izin SELECT pada semua tabel yang dilangganan untuk snapshot awal.

• Hanya terhubung ke primary endpoint kluster PolarDB. Replikasi logis tidak didukung pada cluster endpoint.

• Atur parameter connector.class ke io.debezium.connector.postgresql.PolarDBOConnector.

• Kami merekomendasikan Anda mengatur parameter plugin.name ke pgoutput. Jika tidak, parsing inkremental dapat menghasilkan teks rusak (garbled) untuk database yang tidak menggunakan encoding UTF-8. Untuk informasi lebih lanjut, lihat dokumentasi komunitas.

Contoh

Contoh berikut menjelaskan cara menggunakan konektor Debezium PolarDBO untuk menyinkronkan tabel t1 dan t2 dari database dbz_db di kluster PolarDB for PostgreSQL dengan Oracle compatibility 2.0 ke antrian pesan Kafka.

Prasyarat

1. Siapkan Kafka

   1. Deploy sebuah instance Kafka dan pastikan dapat diakses dari host Kafka Connect. Anda juga dapat menggunakan ApsaraMQ for Kafka. Untuk informasi lebih lanjut, lihat Quick start.

   2. Buat topik bernama pg_dbz_event di instance Kafka untuk menerima pesan.

      Catatan

      Untuk pengujian, Anda dapat membuat topik dengan satu partition agar lebih mudah dilihat. Untuk lingkungan produksi, buat topik multi-partisi.

2. Jalankan Kafka Connect secara lokal dalam distributed mode pada port 8083.

   • Salin paket JAR Debezium PolarDBO connector ke direktori plugin.path Kafka Connect.

     # Ganti ${plugin.path} dengan path aktual.
     mkdir ${plugin.path}/debezium-connector-polardbo
     cp debezium-connector-postgres-polardbo-v1.0-2.6.2.Final.jar ${plugin.path}/debezium-connector-polardbo

3. Siapkan PolarDB for PostgreSQL (Compatible with Oracle)

   1. Di halaman pembelian kluster PolarDB, beli kluster PolarDB for PostgreSQL (Compatible with Oracle) versi 2.0.

   2. Konfigurasikan kluster PolarDB agar memenuhi semua prasyarat yang tercantum di bagian Penggunaan.

   3. Buat akun istimewa. Untuk informasi lebih lanjut, lihat Buat akun.

   4. Dapatkan primary endpoint kluster. Untuk informasi lebih lanjut, lihat Lihat titik akhir koneksi. Jika kluster PolarDB dan instance Kafka Connect berada di availability zone yang sama, Anda dapat menggunakan private endpoint. Jika tidak, Anda harus meminta public endpoint. Tambahkan alamat instance Kafka Connect ke cluster whitelist PolarDB. Untuk informasi lebih lanjut, lihat Konfigurasi daftar putih kluster.

   5. Di Konsol, buat database bernama dbz_db. Untuk informasi lebih lanjut, lihat Buat database.

   6. Jalankan pernyataan berikut untuk membuat tabel t1 dan t2 di database dbz_db dan memasukkan data.

      CREATE TABLE public.t1 (a int PRIMARY KEY, b text, c TIMESTAMP);
      ALTER TABLE public.t1 REPLICA IDENTITY FULL;
      INSERT INTO public.t1(a, b, c) VALUES(1, 'a', now());
      CREATE TABLE public.t2 (a int PRIMARY KEY, b text, c DATE);
      ALTER TABLE public.t2 REPLICA IDENTITY FULL;
      INSERT INTO public.t2(a, b, c) VALUES(1, 'a', now());

Tes

1. Buat file konfigurasi bernama config/postgresql-connector.json. Untuk deskripsi parameter, lihat dokumentasi komunitas.

   {
     "name": "dbz-polardb",
     "config": {
       "connector.class": "io.debezium.connector.postgresql.PolarDBOConnector",
       "database.hostname": "<yourHostname>", 
       "database.port": "<yourPort>", 
       "database.user": "<yourUserName>", 
       "database.password": "<yourPassWord>", 
       "database.dbname" : "dbz_db",
       "plugin.name": "pgoutput",
       "slot.name": "dbz_polardb",
       "table.include.list": "public.t1,public.t2",
       "topic.prefix": "polardb"
       "transforms": "Combine",
       "transforms.Combine.type": "io.debezium.transforms.ByLogicalTableRouter",
       "transforms.Combine.topic.regex": "(.*)",
       "transforms.Combine.topic.replacement": "pg_dbz_event"
     }
   }

   Catatan

   Secara default, Debezium membuat satu topik per tabel. Konfigurasi ini mengarahkan semua event perubahan ke satu topik tunggal.

2. Tambahkan konektor.

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 'http://localhost:8083/connectors' -d @config/postgresql-connector.json

   Setelah konektor ditambahkan, data lengkap tersedia di topik Kafka.

   Di tab message query instance Kafka Anda, atur metode kueri ke query by offset, pilih partisi 0, dan atur start offset ke 0. Lalu, klik Query. Hasilnya menunjukkan dua pesan CDC Debezium (pada offset 0 dan 1). Kunci berisi nilai __dbz__physicalTableIdentifier masing-masing polardb.public.t1 dan polardb.public.t2. Hal ini mengonfirmasi bahwa snapshot data lengkap telah disinkronkan ke Kafka.

3. Jalankan pernyataan DML berikut di database dbz_db kluster PolarDB:

   INSERT INTO public.t1(a, b, c) VALUES(2, 'b', now());
   UPDATE public.t1 SET b = 'c' WHERE a = 1;
   DELETE FROM public.t1 WHERE a = 2;
   INSERT INTO public.t1(a, b, c) VALUES(4, 'd', now());
   INSERT INTO public.t2(a, b, c) VALUES(2, 'b', now());
   UPDATE public.t2 SET b = 'c' WHERE a = 1;
   DELETE FROM public.t2 WHERE a = 2;
   INSERT INTO public.t2(a, b, c) VALUES(4, 'd', now());

   Data inkremental kini tersedia di topik Kafka.

   Di halaman kueri pesan, atur metode kueri ke query by offset, pilih partisi target dan offset awal, lalu klik Query. Hasilnya menunjukkan bahwa Debezium telah menangkap operasi DML pada tabel t1 dan t2 sebagai pesan CDC. Kunci setiap pesan mencakup pengenal tabel (seperti polardb.public.t1) dan nilai primary key. Nilai pesan berisi event perubahan dalam format Debezium, termasuk bidang before dan after. Untuk operasi DELETE, nilai pesan berukuran 0 byte.