PostgreSQL实时同步

本文提供PostgreSQL实时同步场景的实施思路、配置示例和验证方式，可作为实际项目配置参考。

实现在PostgreSQL中结果表向目标表中实时同步数据

案例

PostgreSQL中将cdc_source中的数据实时同步到cdc_target中

前提

• 启动PostgreSQL的logical

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

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

部分PostgreSQL环境需要安装decoderbufs

apt-get install -y postgresql-14-decoderbufs

解决方案

创建Flink SQL作业类型，添加以下依赖

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

来源表和目标表都需要指定主键

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;