如何在Apache Flink中连接PostgreSQL？数据读取异常排查

一、Ververica CDC连接器无数据输出问题 #

排查与修复步骤： #

1. 校验PostgreSQL CDC基础配置

   • 确认PostgreSQL的wal_level已设置为logical（修改postgresql.conf后重启服务），这是CDC功能的核心前提。
   • 确保数据库账号拥有REPLICATION权限，以及目标表的SELECT权限。
   • 修正tableList格式：必须指定完整的库名.模式名.表名，比如默认public模式下要写成database.public.table，仅写表名会导致连接器无法定位目标表。

2. 确认输出日志位置

   • Flink的print()默认输出到TaskManager的stdout日志，而非JobManager日志，需到对应TaskManager节点查看日志。
   • 可替换为printToErr()或添加自定义日志输出，便于快速验证：

     env.addSource(postgresSourceBuilder)
        .map(record -> {
            System.err.println("CDC接收数据: " + record);
            return record;
        })
        .print()
        .setParallelism(1);
     

3. 版本兼容性检查

   • 确保Ververica CDC连接器版本与Flink版本匹配（如Flink 1.17对应Ververica CDC 2.4.x系列），版本不兼容会导致隐性连接或序列化问题。

二、JDBC连接器读取新增数据与Java执行查询 #

1. 实现流式读取新增数据 #

JDBC连接器默认是批模式，仅读取一次历史数据。要实现实时读取新增数据，需配置增量扫描参数：

CREATE TABLE MyTest (
  id INT,
  FIRST NUMERIC(6),
  LAST NUMERIC(6),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:postgresql://ip/database',
   'table-name' = 'table',
   'username'='user',
   'password'='password',
   'scan.incremental.column.name' = 'id', -- 指定自增ID或时间戳作为增量列
   'scan.startup.mode' = 'latest-offset', -- 从当前最大ID开始读取新增数据
   'poll-interval' = '5s' -- 每隔5秒轮询一次新增数据
);

若需要先读取历史数据再实时监控新增，可将scan.startup.mode设为earliest-offset。

2. Java应用中执行SQL查询的实现 #

使用Flink Table API完成表注册与查询：

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class PostgresJDBCStreamReader {
    public static void main(String[] args) {
        // 初始化流式TableEnvironment
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        // 注册JDBC表
        String createTableSQL = "CREATE TABLE MyTest (" +
                "  id INT," +
                "  FIRST NUMERIC(6)," +
                "  LAST NUMERIC(6)," +
                "  PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                "   'connector' = 'jdbc'," +
                "   'url' = 'jdbc:postgresql://ip/database'," +
                "   'table-name' = 'table'," +
                "   'username'='user'," +
                "   'password'='password'," +
                "   'scan.incremental.column.name' = 'id'," +
                "   'scan.startup.mode' = 'latest-offset'," +
                "   'poll-interval' = '5s'" +
                ")";
        tableEnv.executeSql(createTableSQL);

        // 执行查询并输出结果
        tableEnv.executeSql("SELECT * FROM MyTest").print();
    }
}

3. 实现重复读取数据 #

• 全量+增量结合：将scan.startup.mode设为earliest-offset，连接器会先读取全量历史数据，之后自动持续读取新增数据。
• 定期全量扫描：自定义Source或使用Flink的定时器，定期触发全量表扫描，适合需要周期性同步全量数据的场景。
• 断点续读：通过Flink Savepoint机制，任务重启后从上次停止的位置继续读取，避免重复消费数据。

内容的提问来源于stack exchange，提问作者Bzelga