Postgres CDC（公测中）

背景信息

Postgres CDC可用于依次读取PostgreSQL数据库全量快照数据和变更数据，保证不多读一条也不少读一条数据。即使发生故障，也能采用Exactly Once方式处理。

前提条件

• 已创建Postgres数据库和表，创建RDS Postgres数据库和表的详情请参见创建账号和数据库。

• 已在阿里云RDS PostgreSQL、Amazon RDS PostgreSQL或者自建PostgreSQL上进行了相应的配置，详情请参见配置Postgres。

使用限制

• 仅Flink计算引擎VVR 2.0及以上版本支持Postgres的CDC连接器。

• Postgres CDC暂不支持在全表扫描阶段执行Checkpoint。

  如果您的作业在全表扫描阶段触发Checkpoint，则可能由于Checkpoint超时导致作业Failover。因此，建议您在其他配置中配置如下参数，具体操作请参见如何配置作业运行参数？。避免在全量同步阶段由于Checkpoint超时导致Failover。

  execution.checkpointing.interval: 10min
  execution.checkpointing.tolerable-failed-checkpoints: 100
  restart-strategy: fixed-delay
  restart-strategy.fixed-delay.attempts: 2147483647

  相关的参数说明详情如下表所示。

  参数	说明	备注
  execution.checkpointing.interval	Checkpoint的时间间隔。	单位是Duration类型，例如10min或30s。
  execution.checkpointing.tolerable-failed-checkpoints	容忍Checkpoint失败的次数。	该参数的取值与Checkpoint调度间隔时间的乘积就是允许的快照读取时间。 说明 如果表特别大，建议将该参数值配置得大一些。
  restart-strategy	重启策略。	参数取值如下： fixed-delay：固定延迟重启策略。 failure-rate：故障率重启策略。 exponential-delay：指数延迟重启策略。 详情请参见Restart Strategies。
  restart-strategy.fixed-delay.attempts	固定延迟重启策略下，尝试重启的最大次数。	无。

语法结构

CREATE TABLE postgrescdc_source (
  shipment_id INT,
  order_id INT,
  origin STRING,
  destination STRING,
  is_arrived BOOLEAN
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<yourHostname>',
  'port' = '5432',
  'username' = '<yourUserName>',
  'password' = '<yourPassWord>',
  'database-name' = '<yourDatabaseName>',
  'schema-name' = '<yourSchemaName>',
  'table-name' = '<yourTableName>'
);

WITH参数

类型映射

使用示例

CREATE TABLE source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>'
);

SELECT * FROM source;