Flink CDC读取pg数据库，每次能增量读取吗，怎么设置？

Flink CDC读取PostgreSQL（pg）数据库时，支持增量读取，但需要正确配置相关参数以确保增量数据能够被准确捕获。以下是详细的设置方法和注意事项：

1. 增量读取的基本机制

Flink CDC通过PostgreSQL的逻辑复制功能（Logical Replication）读取WAL（Write-Ahead Logging）日志来捕获增量数据。在完成全量数据读取后，Flink CDC会自动切换到增量读取阶段，从而实现流批一体的数据同步。

• 全量读取：读取表中的历史数据。
• 增量读取：从WAL日志中捕获新增、更新或删除的操作。

2. 配置步骤

2.1 数据库端配置

为了支持逻辑复制，您需要对PostgreSQL数据库进行以下配置： 1. 启用逻辑复制： - 修改postgresql.conf文件，设置以下参数： conf wal_level = logical max_replication_slots = 10 # 根据实际需求调整 max_wal_senders = 10 # 根据实际需求调整 - 重启PostgreSQL服务以使配置生效。

1. 创建逻辑复制槽（Replication Slot）：

   • 使用SQL命令创建一个逻辑复制槽，例如：

     SELECT * FROM pg_create_logical_replication_slot('flink_slot', 'pgoutput');
     

   • 注意：逻辑复制槽会保留WAL日志，因此需要定期清理未使用的槽以避免磁盘空间耗尽。

2. 授权用户权限：

   • 确保用于连接的数据库用户具有REPLICATION权限，并对目标表有SELECT权限。

2.2 Flink CDC连接器配置

在Flink作业中，使用Postgres CDC连接器时，需配置以下关键参数：

1. 基本连接参数：

   • connector: 设置为postgres-cdc。
   • hostname: PostgreSQL实例的主机地址。
   • port: PostgreSQL实例的端口号。
   • username: 数据库用户名。
   • password: 数据库密码。
   • database-name: 数据库名称。
   • schema-name: 模式名称。
   • table-name: 表名称。

2. 增量读取相关参数：

   • slot.name: 指定逻辑复制槽的名称，例如flink_slot。
   • heartbeat.interval.ms: 发送心跳包的时间间隔，默认为30s。当表变更不频繁时，设置该值可以及时推进Slot的偏移量。
   • scan.incremental.snapshot.chunk.key-column: 指定某一列作为快照阶段切分分片的切分列，默认从主键中选择第一列。
   • scan.incremental.close-idle-reader.enabled: 是否在快照结束后关闭空闲的Reader，默认为false。
   • scan.incremental.snapshot.backfill.skip: 是否跳过全量阶段的日志读取：
     • true：跳过全量阶段日志读取，增量阶段从低水位线开始读取日志。适用于下游算子或存储支持幂等性的情况。
     • false：不跳过全量阶段日志读取，保证一致性。适用于需要聚合、关联等操作的场景。

3. 示例配置：

   CREATE TABLE postgres_source (
      id INT,
      name STRING,
      age INT,
      PRIMARY KEY (id) NOT ENFORCED
   ) WITH (
      'connector' = 'postgres-cdc',
      'hostname' = 'your-postgres-host',
      'port' = '5432',
      'username' = 'your-username',
      'password' = 'your-password',
      'database-name' = 'your-database',
      'schema-name' = 'public',
      'table-name' = 'your-table',
      'slot.name' = 'flink_slot',
      'heartbeat.interval.ms' = '30000',
      'scan.incremental.snapshot.chunk.key-column' = 'id',
      'scan.incremental.close-idle-reader.enabled' = 'true',
      'scan.incremental.snapshot.backfill.skip' = 'false'
   );
   

3. 注意事项

1. 全量到增量的切换：

   • 在全量读取完成后，Flink CDC会自动切换到增量读取阶段。如果全量读取是多并发的，则需要等待一个Checkpoint完成，以确保全量数据已写入下游后再开始增量读取。
   • 建议：根据业务需求设置合理的Checkpoint间隔时间，避免因间隔过长导致增量数据延迟。

2. 逻辑复制槽管理：

   • 如果Flink作业长时间未运行，逻辑复制槽可能会导致WAL日志堆积，占用大量磁盘空间。建议定期监控并清理未使用的复制槽。

3. 备库限制：

   • Flink CDC不支持从PostgreSQL备库读取增量数据，因为备库无法提供完整的WAL日志。

4. Exactly Once语义：

   • Flink CDC通过Checkpoint机制保证Exactly Once语义。如果作业发生Failover，可以从上次的Checkpoint恢复，确保数据一致性。

4. 总结

通过正确配置PostgreSQL数据库和Flink CDC连接器，您可以实现PostgreSQL数据的全量和增量读取。增量读取依赖于逻辑复制槽和WAL日志，因此需要特别关注数据库的配置和资源管理。合理设置Flink CDC参数（如slot.name、heartbeat.interval.ms等）可以优化增量读取的性能和稳定性。