AnalyticDB:Write data to AnalyticDB for PostgreSQL using Realtime Compute for Apache Flink

Limitations

• Realtime Compute for Apache Flink cannot read data from AnalyticDB for PostgreSQL in serverless mode.

• Only Ververica Runtime (VVR) 6.0.0 or later supports the AnalyticDB for PostgreSQL connector.

• Only VVR 8.0.1 or later supports AnalyticDB for PostgreSQL V7.0.

Prerequisites

Before you begin, ensure that you have:

• A fully managed Flink workspace. See Activate fully managed Flink.

• An AnalyticDB for PostgreSQL instance. See Create an instance.

• The AnalyticDB for PostgreSQL instance and the fully managed Flink workspace in the same virtual private cloud (VPC).

Configure the AnalyticDB for PostgreSQL instance

1. Log on to the AnalyticDB for PostgreSQL console.

2. Add the CIDR block of the fully managed Flink workspace to the IP address whitelist of the instance. See Configure an IP address whitelist.

3. Click Log On to Database. See Client connection for connection options.

4. Create the destination table:

   CREATE TABLE test_adbpg_table(
     b1 int,
     b2 int,
     b3 text,
     PRIMARY KEY(b1)
   );

Set up the Flink connector

The AnalyticDB for PostgreSQL connector is distributed as a JAR file hosted on GitHub. Download the file and upload it to your Flink workspace as a custom connector.

1. Log on to the Realtime Compute for Apache Flink console. On the Fully Managed Flink tab, find your workspace and click Console in the Actions column.

2. In the left-side navigation pane, click Connectors.

3. Click Create Custom Connector and upload the JAR file.

   Note Get the JAR file from GitHub releases. The JAR version must match the Flink engine version of your workspace.

4. Click Next. The system parses the JAR content. If parsing fails, check that the connector code conforms to Apache Flink community standards.

5. Click Finish. The connector appears in the connector list.

Create a Flink job

1. Log on to the Realtime Compute for Apache Flink console. On the Fully Managed Flink tab, find your workspace and click Console in the Actions column.

2. In the left-side navigation pane, click SQL Editor. In the upper-left corner, click New.

3. In the New Draft dialog box, on the SQL Scripts tab, click Blank Stream Draft and click Next.

4. Configure the draft and click Create.

   Parameter	Description	Example
   Name	A unique name for the draft within the current project	adbpg-test
   Location	The folder where the code file is saved	Draft
   Engine Version	The Flink engine version. See Engine version for supported versions	vvr-6.0.7-flink-1.15

Write data to AnalyticDB for PostgreSQL

-- Source: generates random data using the built-in datagen connector
CREATE TABLE datagen_source (
  f_sequence INT,
  f_random INT,
  f_random_str STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '5',
  'fields.f_sequence.kind' = 'sequence',
  'fields.f_sequence.start' = '1',
  'fields.f_sequence.end' = '1000',
  'fields.f_random.min' = '1',
  'fields.f_random.max' = '1000',
  'fields.f_random_str.length' = '10'
);

-- Sink: writes data to AnalyticDB for PostgreSQL via the adbpg connector
CREATE TABLE test_adbpg_table (
  `B1` bigint,
  `B2` bigint,
  `B3` VARCHAR,
  `B4` VARCHAR,
  PRIMARY KEY(B1) NOT ENFORCED
) WITH (
  'connector' = 'adbpg-nightly-1.13',
  'url' = 'jdbc:postgresql://<internal-endpoint>:5432/<database-name>',
  'tablename' = 'test_adbpg_table',
  'username' = '<username>',
  'password' = '<password>',
  'maxretrytimes' = '2',
  'batchsize' = '50000',
  'connectionmaxactive' = '5',
  'conflictmode' = 'ignore',
  'usecopy' = '0',
  'targetschema' = 'public',
  'exceptionmode' = 'ignore',
  'casesensitive' = '0',
  'writemode' = '1',
  'retrywaittime' = '200'
);

Verify the result

1. Connect to the AnalyticDB for PostgreSQL database. See Client connection.

2. Query the table to confirm data is being written:

   SELECT * FROM test_adbpg_table;

   Data is written to the AnalyticDB for PostgreSQL database. The following figure shows the returned result.

   

What's next

• AnalyticDB for PostgreSQL connector — full connector reference, including all parameters and data type mappings

• Datagen connector — reference for the datagen source used in this example

References

• Datagen