说实话，刚开始听说要把 Flink‑CDC 搞到人大金仓，我还真是有点懵。

哈哈，别急，咱慢慢聊，你会发现这事儿其实挺有戏的。

先说背景，干嘛要搬砖

Zui近项目上硬是要把 MySQL 那老古董给换成国产的金仓数据库。

那个那个，迁移不光是把表搬过去，还得保证实时同步。

你懂的，业务数据一秒钟dou不Neng停。

于是就想到了 Flink‑CDC，这玩意儿本来是给 MySQL、PostgreSQL、Oracle 准备的。

可是金仓官方竟然没啥现成的 CDC 包，哎呀妈呀，这不就尴尬了。

一步步拆解难题

先把 MySQL 的全量数据搬过去，用金仓自带的 KDTS 工具。

KDTS 那玩意儿支持 Web 版，也Neng在 Linux 上跑，直接一键导。

别忘了把迁移日志记下来以防哪天翻车还Neng回滚。

接下来就是实时增量同步啦。

我翻遍了 Flink 和金仓的文档，一句话：没有官方方案。

不对不对，我应该说：官方根本没给出示例代码。

所以只Neng动脑子，从兼容性入手——金仓底层是基于 PostgreSQL 开发的！

为什么Ke以套用 PostgreSQL‑CDC？

金仓也支持 WAL逻辑复制，这跟 PostgreSQL 一模一样。

它实现了大部分 PostgreSQL 的协议和驱动兼容层。

换句话说只要我们把 Flink‑CDC 的 PostgreSQL 连接器塞进去，就Neng“骗”它工作。

哈哈，这招叫“借力打力”，在技术圈里算是小技巧吧，你懂的。

准备工作：装驱动

# 在 ${FLINK_HOME}/lib 下添加 PostgreSQL 驱动包

mv /ops/app/postgresql-42.xx.jar ${FLINK_HOME}/lib/

# 在 ${DINKY_HOME}/lib 下添加同样的驱动包

mv /ops/app/postgresql-42.xx.jar ${DINKY_HOME}/customJar/

装完后记得重启 Flink 集群和 Dinky 控制台，不然新加的 jar 不会被加载。

配置金仓 WAL 参数

打开金仓的数据目录下的 kingbase.conf（路径类似 /opt/Kingbase/ES/V8/data/kingbase.conf)。

wal_level = logical   # 把默认改成 logical 才NengZuo逻辑复制

改完记得重启服务：

docker restart kingbase   # 或者系统服务方式自行重启

再跑个 SQL kankan：

SHOW wal_level;

Dinky 中写 Flink‑SQL 脚本

Dinky 打开新建任务，粘贴下面这段代码就行。记住每行dou是独立的

, 让页面kan着geng舒服。

定义源表

CREATE TABLE source_table (
    `id` STRING COMMENT '主键id',
    `created_time` BIGINT NOT NULL COMMENT '创建时间',
    `remark` STRING,
    PRIMARY KEY  NOT ENFORCED
) WITH (
    'connector' = 'postgres-cdc',
    'hostname' = '192.168.1.100',
    'port' = '5432',
    'username' = 'kingbase',
    'password' = '******',
    'database-name' = 'kingbase',
    'schema-name' = 'public',
    'table-name' = 'iot_dict_info',
    -- slot.name 必须唯一，随便起个名字就行
    'slot.name' = 'kc_cdc_slot_source_table',
    -- 初始同步从头开始
    'scan.startup.mode' = 'initial',
    -- 

定义目标表

CREATE TABLE target_table (
    `id` STRING COMMENT '主键id',
    `created_time` BIGINT NOT NULL COMMENT '创建时间',
    `remark` STRING,
    PRIMARY KEY  NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:postgresql://192.168.1.100:5432/kingbase',
    'username' = 'kingbase',
    'password' = '******',
    -- 注意这里写的是目标表全名
    'table-name' = 'public.iot_dict_info2',
    'driver' = 'org.postgresql.Driver'
);

写入同步语句

INSERT INTO target_table
SELECT * FROM source_table;

调试小技巧，你懂的

先单独跑一次 SELECT * FROM source_table kankan是否Neng读到数据；Ru果报错，大概率是 WAL 没开或者 slot 名冲突。

CDC 的 slot 是持久化在数据库里的，一旦创建后即使任务挂掉也会继续保留进度。别忘了清理不需要的 slot，否则会占磁盘空间哦！

Dinky 日志里经常会出现 “Can’t find table …”，那通常是 schema 名写错或者权限不足。检查一下用户有没有 REPLICATION 权限吧！

CDC 同步延迟一般在几百毫秒到几秒之间，别期望毫秒级实时那是奢望啦~

坑点速递——别踩雷！

1️⃣ 金仓默认只开启物理复制，需要手动改成 logical 才Neng用 CDC；这点经常被忽视，我第一次就是忘了改配置文件。

2️⃣ slot.name 必须唯一，同一个库里不Neng两个任务共用同一个 slot，要是不小心用了相同名字，后面的任务会报 “slot already exists”。

3️⃣ Ru果业务库里有大量大对象，一定打开 ‘debezium.lob.enabled’=‘true’，否则同步时会丢失二进制数据。

4️⃣ 金仓对大小写敏感，Ru果你在 MySQL 那边用了全大写表名，要在 PostgreSQL‑CDC 配置里对应好 schema-name 与 table-name 的大小写，否则查询不到。

P.S. 小结一下——怎么玩转 Flink‑CDC 与 人大金仓？

# 步骤一： 使用 KDTS 把全量数据搬过去；确保库结构一致。

# 步骤二： 打开 goldbank.conf，把 wal_level 调为 logical；重启服务。

# 步骤三： 在 Flink/lib 和 Dinky/lib 加入 PostgreSQL JDBC 驱动；重启两套服务。

# 步骤四： 在 Dinky 写 Flink‑SQL，用 ‘postgres-cdc’ connector 指向金仓；记得填好 slot.name 与 schema-name。

# 步骤五： 跑通后监控 lag 与 slot 大小，定期清理不用的 slot 防止磁盘爆满。

# 步骤六： 业务上线前Zuo一次压测，kan下峰值吞吐和延迟是否满足 SLA。

end of story——祝你玩得开心！😎

Ru果还有啥不明白的地方，就留言呗，我这老友偶尔还Neng掏出点经验给你支支招~ 哈哈，祝项目顺利上线，不要再被“兼容性”卡住啦！​

---