INFO-实时开发通过SQL引用CDC插件实现实时计算

使用场景

使用示例

一、postgres数据库准备

-- 开启逻辑复制
alter system set wal_level = logical;  
-- 设置wal_sender为20（推荐值）
alter system set max_wal_senders = 20;  
-- 设置slot数量为20（推荐值）
alter system set max_replication_slots = 20;

create role demo_admin nologin;  
-- 创建用户
create user dsc_demo with password 'xxx' replication login;
-- 将角色授予用户
grant demo_admin to dsc_demo;
-- 为角色授权
grant usage on schema  public to demo_admin;
grant create on database dsc_demo to demo_admin;
grant select,insert,update,delete on all tables in schema  public to demo_admin;
-- 创建PUBLICATION
create publication cdc_pg_demo for table public.pg_demo_src;
-- 需为cdc_pg_demo增加表时
alter publication cdc_pg_demo add table public.pg_demo_sink;

-- 创建pg src表
create table pg_demo_src (
                             id INT PRIMARY KEY,
                             name varchar(20),
                             event_time TIMESTAMP
);

--创建sink表
create table pg_demo_sink (
                             id INT PRIMARY KEY,
                             name varchar(20),
                             event_time TIMESTAMP
);

--设置表的复制格式为full
alter table   pg_demo_src   replica identity full; 
-- 创建PUBLICATION
create publication cdc_pg_demo for table public.pg_demo_src;
-- 需为cdc_pg_demo增加表时
alter publication cdc_pg_demo add table public.pg_demo_sink;

二、实时开发准备

--SQL
--********************************************************************--
--Author: Simon
--CreateTime: 2026-01-22 17:35:22
--Comment: SQL+CDC插件包直接实时消费源端数据库日志，用于当前未支持数据源场景
--********************************************************************--

-- 映射pg源端表，指定实时消费数据来源表
CREATE TABLE flink_demo_src (
  id INT,
  name STRING,
  event_time TIMESTAMP,
   PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'postgres-cdc',
  'changelog-mode' = 'all',
  'hostname' = 'dsc-demo18.jdlt.163.org',
  'port' = '55432',
  'username' = 'dsc_demo',
  'password' = '加密内容1',
  'database-name' = 'dsc_demo',
  'schema-name' = 'public',
  'table-name' = 'pg_demo_src',
  'decoding.plugin.name' = 'pgoutput',
  'slot.name' = 'flink_pg_cdc_01', 
  'scan.incremental.snapshot.enabled' = 'true',
  'scan.incremental.snapshot.chunk.size' = '8096',
  'scan.snapshot.fetch.size' = '1024',
  'debezium.publication.name' = 'cdc_pg_demo', 
  'debezium.publication.autocreate.mode' = 'disabled'
);

-- 映射pg去向表，消费数据后写入的目标表
CREATE TABLE flink_demo_sink (
  id INT,
  name STRING,
  event_time TIMESTAMP,
   PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc', 
  'url' = 'jdbc:postgresql://dsc-demo18.jdlt.163.org:55432/dsc_demo', 
  'table-name' = 'public.pg_demo_sink',  
  'username' = 'dsc_demo',
  'password' = '加密内容2'
);

insert into flink_demo_sink
select * from flink_demo_src;

作者：Simon