Realtime Compute for Apache Flink:PostgreSQL CDC connector (public preview)

Background information

Features

The PostgreSQL CDC supports the incremental snapshot reading feature in Realtime Compute for Apache Flink that uses Ververica Runtime (VVR) 8.0.6 or later. When the connector is run, it first initiates the full scan phase to read the snapshot of a PostgreSQL database. After the full scan is completed, the connector automatically switches to the incremental capture phase to read write-ahead log (WAL) files of the database. If a failure occurs, the connector recovers data in an exactly-once manner to prevent data duplication or loss.

You can use the connector to create a source table, which provides the following advantages:

• Supports reading full and incremental data to facilitate unified stream and batch processing.

• Supports parallel reading in the full scan phase, which improves reading performance.

• Supports automatic and seamless switching from the full scan phase to the incremental capture phase to reduce resource consumption.

• Supports data resumption in the full scan phase to enhance stability.

• Supports lock-free reading in the full scan phase to ensure normal business operation.

Prerequisites

The PostgreSQL CDC connector uses the logical replication of PostgreSQL to read data changes from ApsaraDB RDS for PostgreSQL, Amazon RDS for PostgreSQL, and self-managed PostgreSQL databases.

After you complete the related configurations, make sure that the following conditions are met:

• The wal_level parameter is set to logical, which specifies that the information required for logical decoding is included in WAL files.

• The REPLICA IDENTITY parameter of the subscribed table is set to FULL to ensure data consistency.

  Note

  REPLICA IDENTITY is a PostgreSQL-specific table-level parameter. The parameter specifies whether to include the previous values of the involved table columns in the INSERT and UPDATE events that are generated by the logical decoding plug-in. For more information, see REPLICA IDENTITY.

• The max_wal_senders and max_replication_slots parameters are set to a value greater than the number of occupied replication slots in the monitored database and the number of required replication slots for the Flink deployment.

• Your database account is assigned a superuser role or granted the LOGIN and REPLICATION permissions. Your database account is also granted the SELECT permission on the subscribed table for full data query.

Limits

• The incremental snapshot reading feature of the PostgreSQL CDC connector requires Realtime Compute for Apache Flink that uses VVR 8.0.6 or later.

• You must manually manage replication slots to avoid disk space waste.

  Replication slots are used to retain WAL segments. If a Flink job is restarted, the job can resume from the most recent checkpoint based on the WAL segments, thereby preventing data loss. If you confirm that a Flink job no longer requires a restart, we recommend that you manually delete the corresponding replication slots to free up the occupied resources. In addition, if the progress marker of a replication slot remains stagnant for a long period of time, PostgreSQL retains the WAL records after the marked position. The retention of unnecessary WAL records may result in a significant increase in disk space usage.

• If you enable the incremental snapshot reading feature of the PostgreSQL CDC connector, you must enable checkpointing for the connector and define a primary key in the source table. When the incremental snapshot reading feature is enabled, multiple temporary replication slots are created to achieve parallel reading in the full scan phase.

  The number of required replication slots in the full scan phase is equal to Number of sources × Parallelism + 1. After the PostgreSQL CDC connector switches from the full scan phase to the incremental capture phase, the system automatically removes temporary replication slots and retains only one replication slot. If you want to limit the number of replication slots used in the full scan phase, you can reduce the parallelism. Take note that this also reduces the reading speed. If the downstream operator or storage supports idempotence, you can use the following configuration to skip log reading in the full scan phase: scan.incremental.snapshot.backfill.skip = true. In this case, the connector provides only at-least-once guarantees. If you use SQL to perform operations such as aggregation and association, we recommend that you do not skip log reading in the full scan phase.

• If you disable the incremental snapshot reading feature, the PostgreSQL CDC connector cannot perform checkpoints during the full scan phase.

  If the feature is disabled and checkpoints are triggered during the full scan phase, the deployment may perform a failover due to a checkpoint timeout. To prevent the issue, we recommend that you add the following configurations to the Other Configuration field in the Parameters section of the Configuration tab. For more information, see Management and operations for workspaces and namespaces.

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

  The following table describes the parameters in the preceding configurations.

  Parameter	Description	Notes
  execution.checkpointing.interval	The interval at which checkpoints are triggered.	Data type: DURATION. Example: 10min and 30s.
  execution.checkpointing.tolerable-failed-checkpoints	The maximum number of checkpoint failures that are allowed.	The product of the value of this parameter and the checkpoint scheduling interval is the maximum duration of the full scan phase. Note If the scanned table is excessively large, we recommend that you specify a sufficiently large value.
  restart-strategy	The restart policy.	Valid values: fixed-delay: The job is restarted with a fixed delay. failure-rate: The job is restarted based on the failure rate. exponential-delay: The job is restarted with an exponential delay. For more information, see Restart Strategies.
  restart-strategy.fixed-delay.attempts	The maximum number of restart attempts when you set the restart-strategy parameter to fixed-delay.	N/A.

Syntax

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>'
);

Parameters in the WITH clause

Data type mappings

Sample code

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;

References

• For information about the connectors that are supported by Realtime Compute for Apache Flink, see Supported connectors.

• For more information about how to write data to a PolarDB for Oracle 1.0 result table, see PolarDB for Oracle 1.0 connector.

• You can use the MySQL connector to read data from or write data to ApsaraDB RDS for MySQL, PolarDB for MySQL, and self-managed MySQL databases. For more information, see MySQL connector.