flink-cdc抓取postgres有demo嘛，通过官方的文档发现跑不起来？

在使用Flink CDC抓取PostgreSQL数据时，如果遇到update数据后无法抓取的问题，可能是由于配置或环境设置不正确导致的。以下是详细的解决方案和示例代码，帮助您排查问题并成功运行Flink CDC抓取PostgreSQL的Demo。

1. 确保PostgreSQL的逻辑复制配置正确

Flink CDC依赖PostgreSQL的逻辑复制功能来捕获变更数据（如INSERT、UPDATE、DELETE）。以下配置是必须的：

1.1 配置wal_level参数

• 在PostgreSQL中，确保wal_level参数设置为logical。

  ALTER SYSTEM SET wal_level = logical;
  SELECT pg_reload_conf();
  

• 检查当前值：

  SHOW wal_level;
  

1.2 设置表的REPLICA IDENTITY

• 对于需要同步的表，确保其REPLICA IDENTITY设置为FULL，以支持UPDATE和DELETE操作的完整旧值记录。

  ALTER TABLE your_table REPLICA IDENTITY FULL;
  

1.3 检查max_wal_senders和max_replication_slots

• 确保max_wal_senders和max_replication_slots的值足够大，以支持Flink作业所需的Slot数量。

  SHOW max_wal_senders;
  SHOW max_replication_slots;
  

1.4 权限配置

• 确保用于连接PostgreSQL的用户具有SUPERUSER权限，或者同时拥有LOGIN和REPLICATION权限，并对目标表具有SELECT权限。

2. Flink CDC连接器的配置与示例代码

以下是一个完整的Flink SQL示例，展示如何使用Flink CDC抓取PostgreSQL的数据：

2.1 创建源表

CREATE TABLE postgrescdc_source (
  id INT,
  name STRING,
  description STRING,
  weight DECIMAL(10, 3),
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<yourHostname>',
  'port' = '5432',
  'username' = '<yourUserName>',
  'password' = '<yourPassWord>',
  'database-name' = '<yourDatabaseName>',
  'schema-name' = '<yourSchemaName>',
  'table-name' = '<yourTableName>',
  'decoding.plugin.name' = 'pgoutput', -- 推荐使用pgoutput插件
  'slot.name' = 'flink_slot', -- 自定义逻辑解码槽名称
  'scan.startup.mode' = 'initial' -- 初始模式：先读全量数据，再读增量数据
);

2.2 查询数据

SELECT * FROM postgrescdc_source;

3. 常见问题排查

3.1 UPDATE数据抓取不到

• 原因：可能是REPLICA IDENTITY未设置为FULL，导致UPDATE操作的旧值无法被逻辑解码插件捕获。
  • 解决方法：执行以下SQL语句，将表的REPLICA IDENTITY设置为FULL：

  ALTER TABLE your_table REPLICA IDENTITY FULL;
  

3.2 数据延迟或丢失

• 原因：可能与Checkpoint配置有关。Flink CDC依赖Checkpoint更新Postgres Slot中的LSN（Log Sequence Number）。
  • 解决方法：确保Flink作业启用了Checkpoint，并合理配置相关参数：

  execution.checkpointing.interval: 60000 # 每60秒触发一次Checkpoint
  execution.checkpointing.tolerable-failed-checkpoints: 3 # 容忍失败的Checkpoint次数
  

3.3 WAL日志不释放

• 原因：可能是Postgres的Slot未正确清理，导致WAL日志堆积。
  • 解决方法：定期检查并清理未使用的Slot：

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

4. 注意事项

• 版本要求：确保使用实时计算引擎VVR 8.0.6及以上版本，因为Postgres CDC增量快照功能仅在此版本及之后支持。
• 插件选择：推荐使用pgoutput插件，它是PostgreSQL 10及以上版本的官方内置插件，性能更优且无需额外安装。
• 脏数据处理：如果遇到非法日期等脏数据，可以通过以下参数跳过或记录脏数据：

  'debezium.event.deserialization.failure.handling.mode' = 'warn'
  

5. 示例运行步骤

1. 准备PostgreSQL环境：

   • 配置wal_level、REPLICA IDENTITY、max_wal_senders等参数。
   • 确保用户权限正确。

2. 创建Flink SQL作业：

   • 使用上述SQL语句创建源表并查询数据。

3. 启动Flink作业：

   • 提交Flink作业到集群，观察日志输出，确认是否能正常捕获INSERT、UPDATE和DELETE操作。

通过以上步骤，您可以成功运行Flink CDC抓取PostgreSQL数据的Demo，并解决UPDATE数据抓取不到的问题。如果仍有疑问，请检查PostgreSQL的日志和Flink作业的运行日志，定位具体错误原因。