企业官网建设流程全解析

做了几年的数据接入和实时计算，我越来越觉得Flink CDC已经成了数据同步场景里绕不开的一个方案。尤其是PostgreSQL，配合Flink CDC，几行SQL就能把业务库的变更实时流式地送到下游，不需要额外部署一堆Agent，也不用手写DataX任务去轮询。这段时间我把整套链路从环境准备到生产踩坑完整梳理了一遍，今天直接把这套实战流程和避坑记录分享出来，想快速上手的朋友可以直接照着抄作业。

这篇内容适合正在做数据库实时同步、准备把PostgreSQL数据接入数据仓库或消息队列的开发者。会用到Docker、Flink SQL和PostgreSQL的基础操作，如果你已经对这几个组件有基本认知，那整个上手过程会非常流畅。我会从方案选型讲到环境搭建，再到核心配置和最终验证，最后把最常见的坑全部列出来，争取让你一次跑通。

1. 为什么选择Flink CDC做PostgreSQL实时同步

1.1 CDC的本质：把数据库日志变成数据流

CDC（Change Data Capture，变更数据捕获）的思路其实很简单，就是让下游系统能感知到数据库里的数据变化，而不需要每次全量拉一遍数据。实现方式有很多种，有的基于触发器，有的基于时间戳轮询，但最优雅、对源库影响最小的方式是解析数据库的事务日志。对PostgreSQL来说，就是基于WAL（Write-Ahead Log，预写日志）机制。

你可以把WAL理解为数据库的“日记”——每次增删改操作发生之前，PostgreSQL都会先把变更记录写到WAL里，用来保证崩溃恢复时的数据一致性。既然这些日志已经记录了每一次数据变化，那CDC工具直接去订阅和解析这份日志就顺理成章了。这样做的好处是实时性极高，数据从提交到被下游感知通常只有毫秒级延迟，而且完全不需要修改业务表结构，也不会给业务数据库带来明显的查询压力。

Flink CDC本质上就是把这套日志解析能力封装成了Flink的连接器。对Flink来说，一个CDC数据源就是一个流，每条记录代表一次数据库变更事件，包括INSERT、UPDATE、DELETE操作，以及变更前后的数据快照。这种统一抽象让后续的数据处理变得非常简单，你不需要关心底层日志格式，直接用SQL就能操作这些变更流。

1.2 Flink CDC相比其他同步方案的优势

早些年做PostgreSQL实时同步，常见的选择是Canal（主要面向MySQL）、Debezium或者自己写程序消费WAL。Debezium其实非常强大，它基于Kafka Connect构建，也是Red Hat在维护的顶级开源项目，Flink CDC底层有一部分能力就是借鉴了Debezium的格式设计。但直接用Debezium的问题在于，你需要额外维护一套Kafka Connect集群，还要处理Topic管理和Schema演进，链路变长之后运维成本直线上升。

Flink CDC最核心的优势是“一体化”。它把数据捕获、数据处理、数据投递整合到了同一个Flink作业里。比如你想把PostgreSQL的表同步到Doris，如果用Debezium，你需要部署Kafka Connect、配置Connector、再写一个Flink作业去消费Kafka里的变更数据，最后写入Doris。而用Flink CDC，一个作业就搞定了，而且可以直接用Flink SQL定义同步逻辑，开发效率完全不在一个量级。

另外Flink CDC天然继承了Flink的流处理能力。实时同步不仅仅是简单的“搬数据”，往往还需要过滤字段、类型转换、维表关联、多流合并。这些在纯同步工具里实现很麻烦，在Flink里就是几个SQL算子的事。我之前做过一个场景，需要把PostgreSQL的订单表和MySQL的用户表实时关联后写入数仓，用Flink CDC加一条JOIN就实现了，换做其他方案简直是灾难。

1.3 一个典型的实时链路长什么样

我要演示的场景是比较常见的一种：PostgreSQL业务库中的一张订单表，实时同步到另一个PostgreSQL分析库（或者Doris、StarRocks等OLAP引擎）。链路非常简单：

业务应用写入PostgreSQL主库，PostgreSQL通过逻辑复制将WAL变更发送给Flink CDC连接器，Flink作业解析并处理这些变更，然后通过JDBC连接器写入目标库。

整个过程不需要业务侧做任何改造，不需要双写，不需要中间表。Flink会先做一次全量快照同步，把表里已有的历史数据搬过去，然后自动无缝切换到增量日志消费，继续同步新产生的变更。这个“全量+增量”的自动衔接是Flink CDC非常受欢迎的原因，实际使用中我基本不需要关心切换的衔接点，框架内部处理得很干净。

2. 环境准备与配套版本选型

2.1 软件版本矩阵才是第一个坑

很多人上手Flink CDC直接被版本搞懵。Flink CDC不是一个独立运行的框架，它是以连接器JAR包的形式存在，需要嵌入到Flink环境里。所以你必须保证Flink CDC连接器版本、Flink版本、PostgreSQL驱动版本、甚至PostgreSQL数据库版本之间是兼容的，否则各种诡异报错都会冒出来。

我这次实战使用的版本组合如下（经过验证可以稳定运行）：

这里我要特别强调一下，建议直接使用flink-sql-connector开头的那种带依赖的JAR包，因为它把Debezium相关的依赖、PostgreSQL驱动等都已经打包进去了，你不需要自己去找一堆依赖往Flink的lib目录里塞。我第一次用的时候贪图方便下载了不带依赖的包，结果启动作业时报了各种ClassNotFoundException，折腾半天。用带依赖的包，一个文件就够。

PostgreSQL的版本也需要注意。Flink CDC的PostgreSQL连接器基于Debezium的PostgreSQL Connector实现，逻辑复制功能在PostgreSQL 9.4之后就支持了，但我在实际使用中发现，10以上的版本兼容性最好，14和15都很稳。如果你的库还是9.x甚至更老的版本，建议先升级再考虑实时同步。

2.2 用Docker快速搭建PostgreSQL测试环境

如果你的本机还没有现成的PostgreSQL环境，用Docker起一个是最省事的方式。我这次实战用Docker Compose管理源库和目标库，配置很简单（实际项目中我用的是docker-compose:postgresql这个镜像模板，里面关于版本号和端口映射的部分我按自己的环境做了调整）：

version: '3' services: postgres-source: image: postgres:14 container_name: pg-source environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres POSTGRES_DB: source_db ports: - "5432:5432" command: - "postgres" - "-c" - "wal_level=logical" - "-c" - "max_replication_slots=10" - "-c" - "max_wal_senders=10" postgres-sink: image: postgres:14 container_name: pg-sink environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres POSTGRES_DB: sink_db ports: - "5433:5432"

注意看，我在源库的启动命令里显式指定了wal_level=logical，这一项非常关键。CDC要读取逻辑变更，必须把WAL级别设置为logical，默认的replica级别是不够的。如果漏掉这个配置，Flink CDC作业启动时会直接报错，提示wal_level配置不正确。

启动容器很简单，在docker-compose.yml所在目录执行docker-compose up -d，然后等待镜像拉取和容器启动完成。用docker ps确认两个容器都处于Up状态，然后用docker exec -it pg-source psql -U postgres -d source_db进入源库，建一张测试表并写入一些初始数据。

2.3 准备Flink运行环境

Flink本地跑不需要复杂的集群部署，用Flink的Standalone模式就够了。到Apache Flink官网下载flink-1.17.x的二进制包，解压后就能用。所谓“flink一定要hdfs”的说法其实要分场景——如果你的同步作业是全量加增量的CDC任务，状态后端用RocksDB并配置本地文件系统或者HDFS都可以；如果是纯本地验证，用默认的配置就能跑，不需要强依赖HDFS。我这里直接采用本地模式，把Flink的StateBackend设为filesystem，检查点目录指向本机路径即可。

解压之后，把下载好的Flink CDC连接器JAR包放到FLINK_HOME/lib目录下。如果你用的是SQL客户端方式，还需要把PostgreSQL JDBC驱动也放到lib目录，因为写入端创建目标表时需要加载驱动。我用的是postgresql-42.5.4.jar，放到lib目录后重启Flink SQL客户端让依赖生效。

Flink下载安装的具体步骤在官方文档里有很详细的说明，跟着做基本不会出问题。有一个小细节是如果本机内存不是特别充裕，建议在conf/flink-conf.yaml里把jobmanager.memory.process.size和taskmanager.memory.process.size调小一点，比如都设置成1024m，否则默认值在开发机上可能撑不住。之前有朋友用2G内存的云主机跑Flink，跑着跑着就OOM，把内存调小后稳定多了。

3. 5分钟跑通核心同步流程

3.1 第一步：PostgreSQL源库开启逻辑复制前置条件

在正式创建同步任务之前，源库还需要配置几个前置条件。直接用psql执行下面几条SQL：

-- 创建用于数据同步的账号，并授予复制权限 CREATE USER flink_user WITH PASSWORD 'flink_pass'; ALTER USER flink_user WITH REPLICATION; GRANT CONNECT ON DATABASE source_db TO flink_user; -- 授予表级权限 GRANT SELECT ON ALL TABLES IN SCHEMA public TO flink_user; ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO flink_user; -- 创建逻辑复制发布（Publication） CREATE PUBLICATION flink_pub FOR TABLE orders;

这里有两件事容易被忽略。第一，CDC连接器读取的是PostgreSQL的逻辑复制流，所以同步账号必须具备REPLICATION权限，否则作业连上去之后拿不到日志数据。第二，在PostgreSQL 10及以上版本，逻辑复制是基于Publication机制的，你必须为需要同步的表创建Publication，Flink CDC才能识别并订阅这些表的变更。如果漏掉这一步，任务启动时会报错说找不到表的复制标识。

权限和Publication配置完成之后，可以先用一条命令验证一下复制槽是否正常，后面排查问题时也常用到：

SELECT * FROM pg_replication_slots;

此时应该没有复制槽，因为Flink作业还没启动。等Flink作业启动后再看，这里会多出一条记录。

3.2 第二步：在Flink SQL中创建源表和目标表映射

Flink SQL连接器的使用方式非常直观，先创建一张映射到PostgreSQL源表的虚拟表，再创建一张映射到目标库的表，最后一行INSERT INTO就能启动同步。抛开命令行的准备工作，5分钟跑通核心流程是完全可行的。

先启动Flink SQL客户端：

$FLINK_HOME/bin/sql-client.sh embedded -l $FLINK_HOME/lib

然后在SQL客户端里执行建表语句：

CREATE TABLE orders_source ( id INT, user_id INT, product_name STRING, amount DECIMAL(10, 2), order_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'postgres-cdc', 'hostname' = 'localhost', 'port' = '5432', 'username' = 'flink_user', 'password' = 'flink_pass', 'database-name' = 'source_db', 'schema-name' = 'public', 'table-name' = 'orders', 'decoding.plugin.name' = 'decoderbufs', 'slot.name' = 'flink_slot' );

有几个参数我需要解释一下。decoding.plugin.name指定了解码WAL日志的插件方式，可选值有decoderbufs、wal2json和pgoutput。其中pgoutput是PostgreSQL 10+原生支持的输出插件，稳定性和兼容性最好，我实际用下来推荐直接选pgoutput。有些老教程会推荐decoderbufs，那是因为当时的历史原因，现在用pgoutput不会有问题。slot.name是逻辑复制槽的名称，需要唯一，如果冲突可以换一个名字。

目标端表定义如下，我用了PostgreSQL的JDBC连接器：

CREATE TABLE orders_sink ( id INT, user_id INT, product_name STRING, amount DECIMAL(10, 2), order_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:postgresql://localhost:5433/sink_db', 'table-name' = 'orders', 'username' = 'postgres', 'password' = 'postgres', 'sink.buffer-flush.max-rows' = '1', 'sink.buffer-flush.interval' = '1s' );

如果你希望源表和目标表的字段名不一致，可以在建表SQL里用别名方式重新映射。需要注意的是，目标表在主键或唯一键冲突时的行为取决于JDBC连接器的实现，默认是INSERT，如果同步的场景里UPDATE操作非常多，建议把sink.buffer-flush.max-rows调小一些，减少批量写入时的延迟。

3.3 第三步：提交同步作业并验证结果

核心语句就一行，直接提交：

INSERT INTO orders_sink SELECT * FROM orders_source;

Flink SQL会把这段逻辑编译成一个流式计算作业，启动后先做全量数据同步，把orders_source表里已有的数据读出来写入orders_sink，然后自动切换到增量模式，持续监听PostgreSQL的WAL变更。

同步作业跑起来后，可以做一个简单的验证。先在源库插入几条新数据：

INSERT INTO orders (user_id, product_name, amount, order_time) VALUES (1001, 'iPhone 15', 6999.00, NOW()), (1002, 'MacBook Pro', 12999.00, NOW()), (1001, 'AirPods Pro', 1899.00, NOW());

然后去目标库查询：

SELECT * FROM orders ORDER BY id;

正常情况下，这几条数据会在一秒内出现在目标库中。再试试UPDATE和DELETE：

UPDATE orders SET amount = 6500.00 WHERE id = 1; DELETE FROM orders WHERE id = 3;

回到目标库查询，你会发现这些变更也完整地同步过来了。UPDATE会变成目标表里的新值，DELETE会把对应主键的数据删掉。这就是CDC和普通批量同步的本质区别，它同步的是“操作”，而不只是“数据结果”。

整个核心流程如果准备工作做得充分，5分钟确实够了。我自己第一次操作的时候，卡在版本匹配和权限配置上花了很久，但一旦这两个前置问题解决，后面的SQL配置和启动作业其实就是几分钟的事，这也是为什么我坚持把版本选型和权限配置放到前面重点讲的原因。

3.4 全量和增量衔接时的时间戳与时区问题

很多人在看到全量数据同步完成后，对增量同步是否就绪会有疑虑。Flink CDC会自动管理这个衔接过程，它会先读取当前WAL日志的位点，然后执行全量快照，快照完成后从记录的位置继续消费增量变更。这期间产生的数据变更不会丢，也不会重复消费。实际效果就是全量和增量是平滑衔接的，中间不需要人工干预。

这里有一个很容易被忽略的细节：时间戳类型。PostgreSQL的TIMESTAMP不带时区信息，而TIMESTAMPTZ带时区信息。Flink SQL里TIMESTAMP(3)映射到PostgreSQL是TIMESTAMP WITHOUT TIME ZONE，TIMESTAMP_LTZ(3)映射带时区类型。如果你在同步过程中发现时间字段差了几个小时，大概率是类型映射没配对，用TIMESTAMP_LTZ来代替TIMESTAMP通常能解决这个问题。

我朋友在生产环境做过一个案例，订单时间字段在源库是TIMESTAMPTZ，建Flink表时用了TIMESTAMP，结果同步到下游数仓的时间全部变成了UTC时间而不是北京时间，凌晨的业务数据全部对不上。排查了大半天才发现是类型映射的问题。所以建表时一定要确认时间字段的映射类型。

4. 常见的坑与排查技巧实录

4.1 PostgreSQL侧配置引发的各种问题

WAL级别未修改。这是遇到最多的报错，任务启动时直接抛异常提示wal_level不满足要求。解决办法很简单，修改postgresql.conf或通过启动参数设置wal_level=logical，然后重启PostgreSQL实例。

需要注意的是，修改wal_level需要重启数据库才能生效，ALTER SYSTEM SET只能改配置不能热加载。如果你在云上使用托管的PostgreSQL，比如阿里云RDS，管理控制台里切换到逻辑复制模式，通常也需要实例重启。生产环境操作前务必评估重启影响窗口。

复制槽不清理导致WAL膨胀。当Flink作业停止或异常退出时，PostgreSQL上的复制槽还会保留。此时WAL日志无法被清理，随着数据变更不断产生，磁盘空间会持续增长。这种情况在业务高峰期尤其危险，我见过一个案例，一个复制槽一周时间吃掉了上百GB的磁盘空间。

排查方式也比较简单，检查pg_replication_slots视图，找到active为false的复制槽，手动删除：

SELECT * FROM pg_replication_slots; SELECT pg_drop_replication_slot('flink_slot');

在Flink作业里合理配置checkpoint和重启策略也能减少复制槽不清理的情况，但最稳妥的方式还是定期巡检。如果你用Flink CDC做长周期任务，建议监控PostgreSQL的WAL目录大小，超过阈值及时告警。

权限不足导致连接失败。如果同步账号没有授予REPLICATION权限，作业启动时会报权限相关的错误。确保账号有REPLICATION属性，并且对要同步的表有SELECT权限。如果表所在的Schema不是public，还需要单独授权：

GRANT USAGE ON SCHEMA your_schema TO flink_user; GRANT SELECT ON ALL TABLES IN SCHEMA your_schema TO flink_user;

4.2 Flink SQL运行阶段的经典报错

找不到合适的驱动或连接器类。这类错误多半是JAR包没放对位置或者版本冲突。检查$FLINK_HOME/lib目录下是否有Flink CDC连接器包和JDBC驱动包，确认版本是兼容的。有时候Flink SQL客户端缓存了旧的依赖，执行:quit退出重启客户端也能解决。

“flink type is datev2, but arrow type is dateday”这类类型不匹配错误。这个报错我见过多次，通常出现在Flink CDC同步PostgreSQL到Doris的场景。PostgreSQL的DATE类型在Flink中被解析为DATE，但Doris的Flink Connector内部映射成了DATEV2，而Arrow的日期类型是DATE32，三者之间存在一个映射断层。解决方法是在写入目标表前显式进行类型转换：

CREATE TABLE doris_sink ( dt STRING, ... ) WITH (...); INSERT INTO doris_sink SELECT CAST(order_time AS STRING), ... FROM orders_source;

或者在Flink表定义中直接把时间字段声明为STRING，从源头避免类型转换。

“flink的jdbc连接器异常”一般源于写入端连接串配置错误。比如目标端url写错了端口，或者目标表不存在，或者用户没有对应权限。这类异常日志比较明显，排查起来相对容易。还有一种情况是目标表的主键和源表不一致，导致UPDATE语义不明确，JDBC连接器会报主键冲突。建目标表时尽量保持主键一致。

4.3 常见问题速查表

4.4 我的一些排查习惯

我做CDC任务排查时，有个固定思路可以分享。遇到问题先确定是源库连接问题、数据读取问题还是写入目标问题，用排除法把范围缩小。我会先看Flink Web UI里的TaskManager日志，很多异常在日志里其实已经写得很直白了。

如果是全量同步阶段报错，大概率是类型映射或者权限问题；如果是增量阶段报错，优先检查复制槽状态和WAL日志是否能正常读取；如果同步任务运行一段时间后出现延迟，而Flink本身CPU和内存都不高，那就要看看目标端的写入瓶颈，比如Doris的导入频率限制或者PostgreSQL的锁竞争。

另外有一个小习惯：每次新建同步任务前，我都会用数据库自带的命令手动确认源库的连接信息。比如先测试PostgreSQL能否正常连通，再确认测试表的权限：

PGPASSWORD=flink_pass psql -h localhost -p 5432 -U flink_user -d source_db -c "SELECT * FROM orders LIMIT 1;"

如果这一步能通过，Flink连接器大概率也不会出问题。很多连接报错其实在数据库侧验证一下就能提前发现。

5. 让同步作业更稳定的一些进阶操作

5.1 关于主键和REPLICA IDENTITY的注意事项

Flink CDC要求同步的表必须有主键，否则无法正确处理UPDATE和DELETE操作。如果源表没有主键，全量同步没问题，但增量阶段对已有数据的UPDATE事件会识别不了。PostgreSQL的逻辑复制机制需要知道每一行的唯一标识。

解决的办法有两个。第一个是给源表补上主键，这在业务库中通常比较难操作。第二个是设置REPLICA IDENTITY FULL，让WAL日志记录整行所有字段的旧值：

ALTER TABLE orders REPLICA IDENTITY FULL;

这样设置之后，即使表没有主键，Debezium也能通过对比整行数据来识别变更。缺点是WAL日志的体积会变大，因为每次UPDATE都会记录整行数据。如果不是磁盘特别紧张，这个方案是可用的。但需要说明的是，Flink CDC的PostgreSQL连接器在语义上更推荐有主键的表，所以我建议源表设计时尽量规范主键。

5.2 检查点配置决定同步语义

Flink CDC的“精确一次”语义依赖检查点（Checkpoint）机制。如果checkpoint没配置好，极端情况下可能出现数据重复或丢失。Flink默认的checkpoint间隔可能不适合实时同步场景，建议在提交作业时显式设置：

SET 'execution.checkpointing.interval' = '10s'; SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE'; SET 'state.backend.type' = 'rocksdb'; SET 'state.checkpoint-storage' = 'filesystem'; SET 'state.checkpoints.dir' = 'file:///tmp/flink-checkpoints';

我建议把checkpoint间隔设置在10秒到30秒之间，太频繁会增加存储压力，太稀疏会导致故障恢复时间变长。状态后端用RocksDB，可以避免大量状态数据占用堆内存导致OOM。

5.3 当单表同步变成整库同步

如果想把整个数据库的所有表都同步过来，不需要为每张表单独建一个Flink作业。Flink PostgreSQL CDC连接器支持通过正则表达式匹配表名，一张源表映射就能覆盖多张物理表。上游的table-name参数可以用正则表达式：

CREATE TABLE all_tables_source ( ... ) WITH ( 'connector' = 'postgres-cdc', 'database-name' = 'source_db', 'schema-name' = 'public', 'table-name' = 'orders|users|products', ... );

但这里要提醒一句，多表同步到一个Flink作业时，所有表的Schema必须完全一致，否则下游处理会非常麻烦。如果表结构差异较大，还是建议拆成多个作业单独同步。

5.4 从PostgreSQL到Doris的实战补充

热词里反复出现的“flink type is datev2, but arrow type is dateday”这个问题，我再展开说一下。这个问题的根源在于Doris的新版本引入了DATEV2类型来替代DATE，而Flink CDC在向Doris写入时，内部用Arrow格式传输数据，Arrow的日期类型是DATE32。三者映射关系不统一，就会出现日期字段无法写入的报错。

解决方法有三种。最直接的是在导入前用SQL做一次CAST，把日期字段转成STRING再写入Doris目标表。还有一种方案是升级Doris Connector到较新的版本，新版本已经适配了DATEV2和Arrow的映射。第三种是调整Doris表的字段类型，把DATEV2改回DATE，但这需要Doris版本支持。实践中，我通常直接在Flink SQL里CAST为STRING，简单有效，对下游数仓的日期处理也没有影响。

6. 写在最后的一点个人体验

做了这么多次Flink CDC同步，我的整体感受是：这个方案的上手门槛确实不高，但要把它用稳、用好，需要对底层机制有一定理解。版本匹配、权限配置、复制槽管理、checkpoint设置，这些都是决定作业能否长期稳定运行的关键细节。千万不要等到作业上线后才开始关注这些问题，前置配置省掉的每一步，后面都可能变成半夜爬起来救火的代价。

我个人在刚开始接触Flink CDC时，最受益的一个做法是先在本地用Docker把整套环境完整跑一遍，包括模拟故障场景，比如杀掉Flink作业再重启、在同步过程中对源表执行大事务、观察checkpoint失败时作业的表现。把这些场景都在本地验证过之后，上生产环境心里就有底了。这个经验后面也一直沿用，每次换新的连接器版本，都会先做一轮本地演练再发布。最后再分享一个小技巧：如果你不确定某个Flink CDC连接器的版本和你的Flink版本是否兼容，最直接的办法是去Flink CDC的GitHub Release页面看对应的文档说明，或者在本地用最小案例验证一下，不要盲目相信网上的教程，版本兼容这种事，实测过了才算数。