Flink SQL CDC 采集 PostgreSQL 分区表：从最新 offset 开始

要从最新的 offset 开始采集 PostgreSQL 分区表，可以使用 Flink SQL CDC（Change Data Capture）功能，并结合使用 flink-sql-connector-postgres 插件。

以下是一个示例的 Flink SQL 查询语句，用于从最新的 offset 开始采集 PG 分区表：

CREATE TABLE source_table (
  id INT,
  name STRING,
  partition_key INT,
  offset INT,
  PRIMARY KEY (id, partition_key)
) PARTITIONED BY (partition_key) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<PG_HOSTNAME>',
  'port' = '<PG_PORT>',
  'database-name' = '<PG_DATABASE>',
  'table-name' = '<PG_TABLE>',
  'username' = '<PG_USERNAME>',
  'password' = '<PG_PASSWORD>',
  'startup-mode' = 'latest-offset'
);

在上述查询语句中，'connector' 参数指定了使用 flink-sql-connector-postgres 插件进行 CDC 采集。'hostname'、'port'、'database-name'、'table-name'、'username'、'password' 参数用于指定连接 PostgreSQL 数据库的相关信息。'startup-mode' 参数设置为 'latest-offset' 表示从最新的 offset 开始采集数据。

请根据实际情况替换尖括号中的 PG_HOSTNAME、PG_PORT、PG_DATABASE、PG_TABLE、PG_USERNAME 和 PG_PASSWORD 参数为实际的 PostgreSQL 数据库连接信息。

通过运行上述查询语句，就可以从最新的 offset 开始采集 PostgreSQL 分区表数据了。