前言

        CDC,Change Data Capture,变更数据获取的简称，使用CDC我们可以从数据库中获取已提交的更改并将这些更改发送到下游，供下游使用。这些变更可以包括INSERT,DELETE,UPDATE等。

用户可以在如下的场景使用cdc：

• 实时数据同步：比如将Postgresql库中的数据同步到我们的数仓中。
• 数据库的实时物化视图。

Postgresql数据库配置

Postgresql参数修改

# 更改wal日志方式为logical
wal_level = logical # minimal, replica, or logical
# 更改solts最大数量（默认值为10），flink-cdc默认一张表占用一个slots
max_replication_slots = 20 # max number of replication slots
# 更改wal发送最大进程数（默认值为10），这个值和上面的solts设置一样
max_wal_senders = 20 # max number of walsender processes
# 中断那些停止活动超过指定毫秒数的复制连接，可以适当设置大一点（默认60s）
wal_sender_timeout = 180s # in milliseconds; 0 disable　　

新建用户并且给用户赋流权限

-- 创建用户 cdc_user
CREATE USER cdc_user PASSWORD 'cdc';-- 授权连接
GRANT CONNECT ON DATABASE postgres TO cdc_user;-- 给用户复制流权限
ALTER ROLE cdc_user replication;-- 授权public的使用
GRANT USAGE ON SCHEMA PUBLIC TO cdc_user;

创建复制槽

---- 创建复制槽
-- wal2json decoder_raw test_decoding pgoutput
select pg_create_logical_replication_slot('cdc_1','wal2json');<