Postgres CDC Source Table

Function

The Postgres CDC source table, that is, Postgres streaming source table, is used to read the full snapshot data and changed data of the PostgreSQL database in sequence. The exactly-once processing semantics is used to ensure data accuracy even if a failure occurs.

Prerequisites

• The PostgreSQL version be 9.6, 10, 11, or 12.
• An enhanced datasource connection with the database has been established, so that you can configure security group rules as required.
  • For details about how to set up an enhanced datasource connection, see Enhanced Datasource Connections in the Data Lake Insight User Guide.
  • For details about how to configure security group rules, see Security Group Overview in the Virtual Private Cloud User Guide.
• In Flink cross-source development scenarios, there is a risk of password leakage if datasource authentication information is directly configured. You are advised to use the datasource authentication provided by DLI.

  For details about datasource authentication, see Introduction to Datasource Authentication.

Precautions

• When you create a Flink OpenSource SQL job, set Flink Version to 1.12 on the Running Parameters tab of the job editing page, select Save Job Log, and set the OBS bucket for saving job logs.
• The PostgreSQL version cannot be earlier than PostgreSQL 11.
• If operations such as update will be performed on the Postgres table, you need to run the following statement in PostgreSQL. Note: Replace test.cdc_order with the actual database and table.

  ALTER TABLE test.cdc_order REPLICA IDENTITY FULL

• Before creating the PostgreSQL CDC source table, check whether the current PostgreSQL contains the default plug-in. You can run the following statement in PostgreSQL to query the current plug-ins:

  SELECT name FROM pg_available_extensions;

  If the default plug-in decoderbufs is not available, you need to set the decoding.plugin.name parameter to specify an existing plug-in in PostgreSQL when creating the PostgreSQL CDC source table.

Syntax

create table postgresCdcSource (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
)
with (
  'connector' = 'postgres-cdc',   
  'hostname' = 'PostgresHostname',
  'username' = 'PostgresUsername',
  'password' = 'PostgresPassword',
  'database-name' = 'PostgresDatabaseName',
  'schema-name' = 'PostgresSchemaName',
  'table-name' = 'PostgresTableName'
);

Parameters

Example

In this example, Postgres-CDC is used to read data from RDS for PostgreSQL in real time and write the data to the Print result table. The procedure is as follows (PostgreSQL 11.11 is used in this example):

1. Create an enhanced datasource connection in the VPC and subnet where PostgreSQL locates, and bind the connection to the required Flink elastic resource pool. For details, see Enhanced Datasource Connections.
2. Set PostgreSQL security groups and add inbound rules to allow access from the Flink queue. Test the connectivity using the PostgreSQL address by referring to Testing Address Connectivity. If the connection is successful, the datasource is bound to the queue. Otherwise, the binding fails.
3. In PostgreSQL, create database flink and schema test.
4. Create table cdc_order in the schema test of database flink in PostgreSQL.

   create table test.cdc_order(
     order_id VARCHAR,
     order_channel VARCHAR,
     order_time VARCHAR,
     pay_amount FLOAT8,
     real_pay FLOAT8,
     pay_time VARCHAR,
     user_id VARCHAR,
     user_name VARCHAR,
     area_id VARCHAR,
     primary key(order_id)
   );

5. Run the following SQL statement in PostgreSQL. If you do not run this statement, an error will be reported when the Flink job is executed. For details, see the error message in FAQ.

   ALTER TABLE test.cdc_order REPLICA IDENTITY FULL

6. Create a Flink OpenSource SQL job. Enter the following job script and submit the job.
   When you create a job, set Flink Version to 1.12 on the Running Parameters tab. Select Save Job Log, and specify the OBS bucket for saving job logs. Change the values of the parameters in bold as needed in the following script.

   create table postgresCdcSource(
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id STRING,
     primary key (order_id) not enforced
   ) with (
     'connector' = 'postgres-cdc',
     'hostname' = 'PostgresHostname',
     'username' = 'PostgresUsername',
     'password' = 'PostgresPassword',
     'database-name' = 'flink',
     'schema-name' = 'test',
     'table-name' = 'cdc_order'
   );
   
   create table printSink(
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id STRING,
     primary key(order_id) not enforced
   ) with (
     'connector' = 'print'
   );
   
   insert into printSink select * from postgresCdcSource;

7. Run the following command in PostgreSQL:

   insert into test.cdc_order
     (order_id,
     order_channel,
     order_time,
     pay_amount,
     real_pay,
     pay_time,
     user_id,
     user_name,
     area_id) values
     ('202103241000000001', 'webShop', '2021-03-24 10:00:00', '100.00', '100.00', '2021-03-24 10:02:03', '0001', 'Alice', '330106'),
     ('202103251202020001', 'miniAppShop', '2021-03-25 12:02:02', '60.00', '60.00', '2021-03-25 12:03:00', '0002', 'Bob', '330110');
   
   update test.cdc_order set order_channel = 'webShop' where order_id = '202103251202020001';
   
   delete from test.cdc_order where order_id = '202103241000000001';

8. Perform the following operations to view the data result in the taskmanager.out file:
   1. Log in to the DLI console. In the navigation pane, choose Job Management > Flink Jobs.
   2. Click the name of the corresponding Flink job, choose Run Log, click OBS Bucket, and locate the folder of the log you want to view according to the date.
   3. Go to the folder of the date, find the folder whose name contains taskmanager, download the taskmanager.out file, and view result logs.

   The data result is as follows:

   +I(202103241000000001,webShop,2021-03-24 10:00:00,100.0,100.0,2021-03-24 10:02:03,0001,Alice,330106)
   +I(202103251202020001,miniAppShop,2021-03-25 12:02:02,60.0,60.0,2021-03-25 12:03:00,0002,Bob,330110)
   -U(202103251202020001,miniAppShop,2021-03-25 12:02:02,60.0,60.0,2021-03-25 12:03:00,0002,Bob,330110)
   +U(202103251202020001,webShop,2021-03-25 12:02:02,60.0,60.0,2021-03-25 12:03:00,0002,Bob,330110)
   -D(202103241000000001,webShop,2021-03-24 10:00:00,100.0,100.0,2021-03-24 10:02:03,0001,Alice,330106)

FAQ

• Q: What should I do if the Flink job execution fails and the log contains the following error information?

  org.postgresql.util.PSQLException: ERROR: logical decoding requires wal_level >= logical

• A: Change the value of wal_level to logical and restart the PostgreSQL database.

  After modifying the PostgreSQL parameter, restart the RDS PostgreSQL instance for the modification to take effect.