Konfigurasi sumber PostgreSQL CDC untuk mengaktifkan Flink real-time streaming - Realtime Compute for Apache Flink - Alibaba Cloud - Realtime Compute for Apache Flink

Sebelum menggunakan konektor PostgreSQL Change Data Capture (CDC), pastikan penyimpanan PostgreSQL hulu memenuhi persyaratan konfigurasi. Topik ini menjelaskan persyaratan konfigurasi dan langkah terkait untuk ApsaraDB RDS for PostgreSQL, Amazon RDS for PostgreSQL, dan database PostgreSQL yang dikelola sendiri.

Latar Belakang

Sebelum menggunakan konektor PostgreSQL CDC, Anda harus memastikan bahwa penyimpanan PostgreSQL hulu memenuhi persyaratan konfigurasi. Topik ini menjelaskan cara mengonfigurasi ApsaraDB RDS for PostgreSQL, Amazon RDS for PostgreSQL, dan database PostgreSQL yang dikelola sendiri. Konektor PostgreSQL CDC melakukan sinkronisasi data berdasarkan mekanisme replikasi logis PostgreSQL. Untuk informasi selengkapnya, lihat Logical Decoding Concepts dan Principles and Best Practices of Logical Subscription.

Database ApsaraDB RDS for PostgreSQL

Koneksi jaringan antara ApsaraDB RDS for PostgreSQL dan Ververica Platform (VVP)
- Jika ApsaraDB RDS for PostgreSQL dan VVP berada dalam virtual private cloud (VPC) yang sama di wilayah yang sama, keduanya sudah terhubung.
- Jika ApsaraDB RDS for PostgreSQL dan VVP ditempatkan di jaringan yang berbeda, Anda harus membuat koneksi jaringan antara keduanya menggunakan metode seperti VPN Gateway. Untuk informasi selengkapnya, lihat Bagaimana Realtime Compute for Apache Flink mengakses Internet? dan Bagaimana Realtime Compute for Apache Flink mengakses layanan lintas VPC?
- Konfigurasikan daftar putih alamat IP untuk ApsaraDB RDS for PostgreSQL agar VVP dapat mengakses ApsaraDB RDS for PostgreSQL. Untuk informasi selengkapnya, lihat Konfigurasikan daftar putih IP.
Konfigurasi server ApsaraDB RDS for PostgreSQL
- Versi yang didukung
  Hanya data dari PostgreSQL 10, PostgreSQL 11, PostgreSQL 12, PostgreSQL 13, PostgreSQL 14, PostgreSQL 15, dan PostgreSQL 16 yang dapat disinkronkan. Anda dapat menjalankan perintah select version() untuk mengecek versi PostgreSQL.
- Logical decoding
  - Konfigurasikan wal_level = logical untuk mengaktifkan logical decoding. Untuk informasi selengkapnya, lihat Ubah parameter instans ApsaraDB RDS for PostgreSQL. Jika konfigurasi berhasil, Anda dapat menguji dan menggunakan fitur ini berdasarkan contoh yang dijelaskan di Logical decoding.
  - Pastikan pengaturan REPLICA IDENTITY pada tabel yang datanya perlu disinkronkan adalah FULL.
    Anda dapat menjalankan pernyataan SQL berikut untuk melihat dan mengubah pengaturan REPLICA IDENTITY.
```
-- Lihat pengaturan REPLICA IDENTITY. 
SELECT CASE relreplident
  WHEN 'd' THEN 'default'
  WHEN 'n' THEN 'nothing'
  WHEN 'f' THEN 'full'
  WHEN 'i' THEN 'index'
END AS replica_identity
FROM pg_class
WHERE oid = 'mytablename'::regclass;

-- Ubah pengaturan REPLICA IDENTITY. 
ALTER TABLE mytablename REPLICA IDENTITY FULL;
```
Plugin sinkronisasi data
Plugin logical decoding wal2json hanya didukung jika versi mesin utama instans ApsaraDB RDS for PostgreSQL adalah PostgreSQL 11, PostgreSQL 12, atau PostgreSQL 13. Plugin pgoutput hanya didukung jika versi mesin utama instans ApsaraDB RDS for PostgreSQL adalah PostgreSQL 10 atau lebih baru. Kami merekomendasikan penggunaan pgoutput dan decodebuf. Untuk informasi selengkapnya, cari opsi decoding.plugin.name pada topik PostgreSQL CDC.
Catatan
Jika Anda menggunakan plugin pgoutput untuk menyinkronkan data, pastikan akun yang digunakan adalah Akun istimewa. Untuk informasi selengkapnya, lihat Buat akun.

Amazon RDS for PostgreSQL

Koneksi jaringan antara Amazon RDS for PostgreSQL dan VVP
Untuk membuat koneksi jaringan antara Amazon RDS for PostgreSQL dan VVP, Anda harus mengizinkan VVP mengakses Amazon RDS for PostgreSQL melalui Internet. Untuk informasi selengkapnya, lihat Bagaimana Realtime Compute for Apache Flink mengakses Internet? dan Controlling access with security groups.
Konfigurasi server Amazon RDS for PostgreSQL
- Versi yang didukung
  Hanya data dari PostgreSQL 10, PostgreSQL 11, PostgreSQL 12, PostgreSQL 13, PostgreSQL 14, PostgreSQL 15, dan PostgreSQL 16 yang dapat disinkronkan. Anda dapat menjalankan perintah select version() untuk mengecek versi PostgreSQL.
- Logical decoding
  - Atur parameter rds.logical_replication ke 1.
  - Pastikan nilai parameter wal_level adalah logical. Secara default, jika parameter rds.logical_replication diatur ke 1, parameter wal_level akan secara otomatis diatur ke logical.
  - Pastikan pengaturan REPLICA IDENTITY pada tabel yang datanya perlu disinkronkan adalah FULL.
    Anda dapat menjalankan pernyataan SQL berikut untuk melihat dan mengubah pengaturan REPLICA IDENTITY.
```
-- Lihat pengaturan REPLICA IDENTITY. 
SELECT CASE relreplident
  WHEN 'd' THEN 'default'
  WHEN 'n' THEN 'nothing'
  WHEN 'f' THEN 'full'
  WHEN 'i' THEN 'index'
END AS replica_identity
FROM pg_class
WHERE oid = 'mytablename'::regclass;

-- Ubah pengaturan REPLICA IDENTITY. 
ALTER TABLE mytablename REPLICA IDENTITY FULL;
```
  - Persyaratan role akun
    Secara default, hanya akun manajemen Amazon Web Services (AWS) yang diberi role rds_replication pada Amazon RDS for PostgreSQL. Pengguna yang diberi role ini dapat mengelola slot logis dan membaca data streaming menggunakan slot logis. Jika Anda ingin mengaktifkan replikasi logis menggunakan akun selain akun manajemen, Anda harus memberikan role rds_replication kepada akun tersebut. Misalnya, Anda dapat memberikan role rds_replication kepada pengguna yang ditentukan oleh <my_user>. Anda hanya dapat memberikan role rds_replication kepada pengguna jika memiliki izin superuser. Jika Anda ingin membuat snapshot awal menggunakan akun selain akun manajemen, Anda harus memberikan izin SELECT pada tabel yang ingin Anda tangkap kepada akun tersebut.
- Plugin sinkronisasi data
  Gunakan plugin pgoutput untuk sinkronisasi data.

Database PostgreSQL yang dikelola sendiri

Koneksi jaringan antara database PostgreSQL yang dikelola sendiri dan VVP
- Jika database PostgreSQL yang dikelola sendiri ditempatkan di Alibaba Cloud dan berada dalam VPC yang sama dengan VVP, koneksi jaringan antara keduanya telah terbentuk.
- Jika database PostgreSQL yang dikelola sendiri tidak ditempatkan di Alibaba Cloud, Anda harus mengizinkan VVP mengakses alamat IP publik database tersebut. Dalam hal ini, Anda harus menghubungkan database PostgreSQL yang dikelola sendiri dan VVP menggunakan VPN Gateway. Untuk informasi selengkapnya, lihat Bagaimana Realtime Compute for Apache Flink mengakses Internet?. Anda juga perlu mengubah file pg_hba.conf untuk mengizinkan VVP mengakses database PostgreSQL yang dikelola sendiri. Untuk informasi selengkapnya, lihat Perbarui file pg_hba.conf.
Konfigurasi server PostgreSQL yang dikelola sendiri
- Versi yang didukung
  Hanya data dari PostgreSQL 9.6, PostgreSQL 10, PostgreSQL 11, PostgreSQL 12, PostgreSQL 13, PostgreSQL 14, PostgreSQL 15, dan PostgreSQL 16 yang dapat disinkronkan. Anda dapat menjalankan perintah select version() untuk mengecek versi PostgreSQL.
- Logical decoding
  - Pastikan nilai parameter wal_level adalah logical. Ubah file postgresql.conf dan restart database.
  - Pastikan pengaturan REPLICA IDENTITY pada tabel yang datanya perlu disinkronkan adalah FULL.
    Anda dapat menjalankan pernyataan SQL berikut untuk melihat dan mengubah pengaturan REPLICA IDENTITY.
```
-- Lihat pengaturan REPLICA IDENTITY. 
SELECT CASE relreplident
   WHEN 'd' THEN 'default'
   WHEN 'n' THEN 'nothing'
   WHEN 'f' THEN 'full'
   WHEN 'i' THEN 'index'
   END AS replica_identity
FROM pg_class
WHERE oid = 'mytablename'::regclass;

-- Ubah pengaturan REPLICA IDENTITY. 
ALTER TABLE mytablename REPLICA IDENTITY FULL;
```
  - Buat akun dan berikan izin kepada akun tersebut untuk menyinkronkan data.
    Akun sinkronisasi data harus memiliki izin REPLICATION dan LOGIN, serta izin SELECT untuk menyinkronkan data tabel.
```
CREATE ROLE <replication_user> REPLICATION LOGIN;

GRANT SELECT ON <table_name> TO <name>;
```
- Plugin sinkronisasi data
  Secara default, plugin pgoutput telah terinstal di PostgreSQL 10 dan versi lebih baru. Jika Anda menggunakan versi PostgreSQL lain atau menginstal plugin lain untuk sinkronisasi data, ikuti petunjuk yang dijelaskan di Logical Decoding Output Plug-in Installation for PostgreSQL.
  Jika Anda menggunakan plugin pgoutput sebagai tool sinkronisasi, pastikan akun sinkronisasi data adalah pemilik tabel yang datanya perlu disinkronkan. Untuk memastikan akun sinkronisasi data menjadi pemilik tabel yang datanya perlu disinkronkan, lakukan langkah-langkah berikut:
```
-- 1. Buat kelompok sinkronisasi. 
CREATE ROLE <replication_group>;

-- 2. Tambahkan pemilik asli tabel ke kelompok sinkronisasi. 
GRANT REPLICATION_GROUP TO <original_owner>;

-- 3. Tambahkan akun sinkronisasi data ke kelompok sinkronisasi. 
GRANT REPLICATION_GROUP TO <replication_user>;

-- 4. Alihkan izin terkait ke kelompok sinkronisasi. 
ALTER TABLE <table_name> OWNER TO REPLICATION_GROUP;
```
  Catatan
  Dalam kasus ini, akun sinkronisasi data hanya menjadi pemilik satu tabel. Oleh karena itu, atur parameter debezium.publication.autocreate.mode dalam klausa WITH ke filtered. Jika proses konfigurasi izin terkait untuk setiap tabel terlalu rumit, Anda dapat langsung memberikan izin pg_monitor kepada akun sinkronisasi data.
  GRANT pg_monitor TO <replication_user>;