ApsaraDB RDS:Gunakan\ slot\ replikasi\ untuk\ menyiapkan\ pelacakan\ perubahan

Langkah 1: Buat database uji

1. Buka halaman Instances. Di bilah navigasi atas, pilih wilayah tempat instans RDS Anda berada. Klik ID instans.

2. Di panel navigasi kiri, klik Databases.

3. Klik Create Database.

   Catatan

   Contoh ini membuat database bernama testdb. Untuk petunjuknya, lihat Buat database.

   

Langkah 2: Buat akun dan konfigurasi izin

1. Di panel navigasi kiri, klik Accounts.

2. Klik Create Account untuk membuat dua akun:

   • Akun istimewa bernama db_admin untuk administrasi

   • Akun standar bernama cdc_user untuk menangkap data perubahan

   Catatan

   Nama akun dalam contoh ini hanya sebagai referensi. Untuk petunjuknya, lihat Buat akun.

3. Lakukan koneksi ke instans sebagai db_admin:

   Catatan

   Untuk menemukan titik akhir instans Anda, lihat Lihat dan ubah titik akhir serta nomor port.

   psql -h <instance-endpoint> -p 5432 -U db_admin -d testdb

4. Berikan role replication kepada cdc_user:

   ALTER USER cdc_user WITH REPLICATION;

   Verifikasi pemberian tersebut:

   SELECT rolreplication FROM pg_roles WHERE rolname='cdc_user';

   Output yang diharapkan:

   rolreplication
   ----------------
    t
   (1 row)

5. Berikan izin SELECT pada semua tabel kepada cdc_user:

   GRANT SELECT ON ALL TABLES IN SCHEMA PUBLIC to cdc_user;

Langkah 3: Atur wal_level menjadi logical

Parameter wal_level mengontrol seberapa banyak informasi yang ditulis ke log WAL. Nilai default-nya adalah replica. CDC memerlukan nilai logical, yang menambahkan informasi yang diperlukan untuk logical decoding.

Untuk memverifikasi nilai saat ini:

SELECT name, setting, short_desc, source
FROM pg_settings
WHERE name = 'wal_level';

Output yang diharapkan:

name      | setting |                          short_desc                          |       source
----------+---------+---------------------------------------------------------------+--------------------
 wal_level | replica | Sets the level of information written to the WAL.            | configuration file
(1 row)

Untuk mengubah wal_level menjadi logical:

1. Buka halaman Instances dan klik ID instans Anda.

2. Di panel navigasi kiri, klik Parameters.

3. Atur wal_level menjadi logical dan kirim perubahan tersebut.

Langkah 4: Buat slot replikasi logis

Tunggu hingga instans restart dan kembali ke status Running sebelum melanjutkan.

1. Lakukan koneksi ke instans sebagai db_admin:

   psql -h <instance-endpoint> -p 5432 -U db_admin -d testdb

2. Buat slot replikasi logis bernama cdc_replication_slot menggunakan plugin output test_decoding:

   Catatan

   test_decoding adalah plugin output yang disediakan oleh PostgreSQL open source. cdc_replication_slot adalah nama contoh — gunakan nama apa pun yang sesuai dengan kebutuhan Anda.

   SELECT pg_create_logical_replication_slot('cdc_replication_slot', 'test_decoding');

   Output yang diharapkan:

   pg_create_logical_replication_slot
   ------------------------------------
    (cdc_replication_slot,1/14003428)
   (1 row)

3. Verifikasi bahwa slot replikasi telah dibuat:

   SELECT * FROM pg_replication_slots;

   Output yang diharapkan:

   slot_name            |    plugin     | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size | two_phase
   ---------------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+---------------+-----------
    cdc_replication_slot | test_decoding | logical   |  18822 | testdb   | f         | f      |            |      |        22356 | 1/140033F0  | 1/14003428          | reserved   |               | f
   (1 row)

Langkah 5: Buat data uji

Jalankan pernyataan berikut untuk membuat tabel uji dan memasukkan data sampel:

CREATE TABLE public.tb_test(
    id int NOT NULL PRIMARY KEY
);

ALTER TABLE public.tb_test ADD name varchar(1) NULL;

INSERT INTO public.tb_test SELECT 1, 'A';

Langkah 6: Baca data perubahan dari slot replikasi

1. Lakukan koneksi ke instans sebagai cdc_user:

   psql -h <instance-endpoint> -p 5432 -U cdc_user -d testdb

2. Baca data perubahan dari slot replikasi tanpa mengonsumsinya:

   SELECT * FROM pg_logical_slot_peek_changes('cdc_replication_slot', null, null);

   Output yang diharapkan:

   lsn        |  xid  |                                  data
   -----------+-------+------------------------------------------------------------------------
    1/14003D90 | 22376 | BEGIN 22376
    1/1401DDE8 | 22376 | COMMIT 22376
    1/1401DDE8 | 22377 | BEGIN 22377
    1/1401E100 | 22377 | COMMIT 22377
    1/1401E2A8 | 22382 | BEGIN 22382
    1/1401E2A8 | 22382 | table public.tb_test: INSERT: id[integer]:1 name[character varying]:'A'
    1/1401E3C0 | 22382 | COMMIT 22382
   (7 rows)

Langkah 7: Konsumsi data perubahan menggunakan pg_recvlogical

pg_recvlogical adalah tool logical decoding bawaan yang disertakan dengan PostgreSQL. Tool ini membaca dari slot replikasi dan memajukan posisi konsumsi sehingga data WAL yang telah dikonsumsi dapat dibersihkan.

1. Jalankan \q untuk menutup koneksi database saat ini.

2. Jalankan perintah berikut untuk mulai mengonsumsi data perubahan:

   Catatan

   Jalankan pg_recvlogical sebagai user postgres (su - postgres). Jika Anda melihat -bash: pg_recvlogical: command not found, lihat FAQ.

   pg_recvlogical -h <instance-endpoint> -U <privileged-account> -d <database> \
     --create-slot --if-not-exists --slot=cdc_replication_slot \
     --plugin=test_decoding --start -f -

   Contoh:

   pg_recvlogical -h pgm-*****.pgsql.singapore.rds.aliyuncs.com -U db_admin -d testdb \
     --create-slot --if-not-exists --slot=cdc_replication_slot \
     --plugin=test_decoding --start -f -

   Output yang diharapkan:

   BEGIN 22376
   COMMIT 22376
   BEGIN 22377
   COMMIT 22377
   BEGIN 22382
   table public.tb_test: INSERT: id[integer]:1 name[character varying]:'A'
   COMMIT 22382