PostgreSQL real-time synchronization

This article provides implementation ideas, configuration examples, and verification methods for PostgreSQL real-time synchronization scenarios, which can be used as a reference for actual project configuration.

Real-time synchronization of data from the result table to the target table in PostgreSQL

Case

Synchronize data in cdc_source to cdc_target in real time in PostgreSQL

Premise

• Start PostgreSQL logical

-- logical 为开启，重启服务生效
ALTER SYSTEM SET wal_level = 'logical';
SHOW wal_level;

-- 来源表一定要设置为FULL，否则更新删除会报错，不需要重启服务
ALTER TABLE public.cdc_source REPLICA IDENTITY FULL;

Some PostgreSQL environments require decoderbufs to be installed

apt-get install -y postgresql-14-decoderbufs

solution

Create a Flink SQL job type and add the following dependencies

• flink-connector-postgres-cdc-3.2.1.jar download
• PostgreSQL-42.7.2.jar download
• flink-cdc-base-3.2.1.jar download
• flink-connector-debezium-3.2.1.jar download
• debezium-connector-postgres-1.9.8.Final.jar download
• debezium-core-1.9.8.Final.jar download
• connect-api-3.2.0.jar download
• debezium-api-1.9.8.Final.jar download
• jackson-databind-2.13.5.jar download
• jackson-core-2.13.5.jar download
• jackson-annotations-2.13.5.jar download
• kafka-storage-3.2.0.jar download
• connect-runtime-3.2.0.jar download
• kafka-clients-3.2.0.jar download
• debezium-embedded-1.9.8.Final.jar download
• connect-JSON-3.2.0.jar download
• HikariCP-4.0.3.jar download
• protobuf-java-3.21.9.jar download

Both the source table and the target table need to specify the primary key

CREATE TABLE from_table
(
    username STRING,
    age      INT,
    PRIMARY KEY (username) NOT ENFORCED
) WITH (
      'connector' = 'postgres-cdc',
      'hostname' = 'localhost',
      'port' = '54302',
      'username' = 'root',
      'password' = 'root123',
      'database-name' = 'isxcode_db',
      'schema-name' = 'public',
      'table-name' = 'cdc_source',
      'slot.name' = 'flink',
      'scan.incremental.snapshot.enabled' = 'true'
      );

CREATE TABLE target_table
(
    username STRING,
    age      INT,
    PRIMARY KEY (username) NOT ENFORCED
) WITH (
      'connector' = 'jdbc',
      'url' = 'jdbc:postgresql://localhost:54302/isxcode_db',
      'driver' = 'org.postgresql.Driver',
      'table-name' = 'cdc_sink',
      'username' = 'root',
      'password' = 'root123',
      'sink.buffer-flush.max-rows' = '1');

INSERT INTO target_table
select *
from from_table;