Write data to AnalyticDB for PostgreSQL-AnalyticDB(AnalyticDB)-阿里云帮助中心

Usage limits

This feature does not support AnalyticDB for PostgreSQL Serverless mode.
Only Realtime Compute for Apache Flink VVR 6.0.0 or later supports the AnalyticDB for PostgreSQL connector.
Only Realtime Compute for Apache Flink VVR 8.0.1 or later supports AnalyticDB for PostgreSQL V7.0.

Note
If you use a custom connector, see Manage custom connectors for more information.

Prerequisites

A Fully Managed Flink workspace is created. For more information, see Activate Realtime Compute for Apache Flink.
An AnalyticDB for PostgreSQL instance is created. For more information, see Create an instance.
The AnalyticDB for PostgreSQL instance and the Fully Managed Flink workspace must be deployed in the same VPC.

Configure an AnalyticDB for PostgreSQL instance

Log on to the AnalyticDB for PostgreSQL console.
Add the CIDR block of the Flink workspace to the whitelist of the AnalyticDB for PostgreSQL instance. For more information about how to configure a whitelist, see Configure a whitelist.
Click Log On to Database. For more information about other ways to connect to the database, see Connect to an instance by using a client.
Create a table on the AnalyticDB for PostgreSQL instance.
Sample SQL statement for creating a table:
```
CREATE TABLE test_adbpg_table(
b1 int,
b2 int,
b3 text,
PRIMARY KEY(b1)
);
```

Configure Realtime Compute for Flink

Log on to the Realtime Compute console.
On the Fully Managed Flink tab, click Console in the Actions column of the target workspace.
In the left-side navigation pane, click Connectors.
On the Connectors page, click Create Custom Connector.
Upload the custom connector JAR file.
Note
- Obtain the JAR package of the custom AnalyticDB for PostgreSQL Flink connector. For more information, see AnalyticDB PostgreSQL Connector.
- The version of the JAR package must match the Flink engine version of the Realtime Compute platform.
After the upload is complete, click Next.

The system parses the uploaded custom connector. If the parsing is successful, you can proceed to the next step. If the parsing fails, check whether the uploaded custom connector code complies with the Apache Flink community standards.
Click Finish.

The created custom connector appears in the connector list.

Create a Flink job

Log on to the Realtime Compute console. On the Fully Managed Flink tab, click Console in the Actions column of the target workspace.
In the left-side navigation pane, click SQL Development. Click New, select Blank Streaming Job Draft, and then click Next.

In the New File Draft dialog box, configure the job parameters.

Job Parameter	Description	Example
Name	The name of the job. Note The job name must be unique within the current project.	adbpg-test
Location	The folder to which the code file of the job belongs. You can also click the icon on the right side of an existing folder to create a subfolder.	Job Drafts
Engine Version	The Flink engine version used by the current job. For more information about engine version numbers, version mappings, and lifecycle milestones, see Engine versions.	vvr-6.0.7-flink-1.15

Click Create.

Write data to AnalyticDB for PostgreSQL

Write the job code.

Create a random source table datagen_source and a destination test_adbpg_table table in AnalyticDB for PostgreSQL. Copy the following job code to the job text editor.

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'
);
CREATE TABLE test_adbpg_table (
    `B1` bigint   ,
    `B2` bigint  ,
    `B3` VARCHAR ,
    `B4` VARCHAR,
     PRIMARY KEY(B1) not ENFORCED
) with (
   'connector' = 'adbpg-nightly-1.13',
   'password' = 'xxx',
   'tablename' = 'test_adbpg_table',
   'username' = 'xxxx',
   'url' = 'jdbc:postgresql://url:5432/schema',
   'maxretrytimes' = '2',
   'batchsize' = '50000',
   'connectionmaxactive' = '5',
   'conflictmode' = 'ignore',
   'usecopy' = '0',
   'targetschema' = 'public',
   'exceptionmode' = 'ignore',
   'casesensitive' = '0',
   'writemode' = '1',
   'retrywaittime' = '200'
);

You do not need to modify the parameters of the datagen_source table. You must modify the parameters of the test_adbpg_table table based on your actual business requirements. The following table describes the parameters.

Parameter	Required	Description
connector	Yes	The name of the connector. Set this parameter to `adbpg-nightly-<version number>`, for example, `adbpg-nightly-1.13`.
url	Yes	The JDBC URL of the AnalyticDB for PostgreSQL instance. Format: `jdbc:postgresql://<internal endpoint>:<port>/<database name>`. Example: `jdbc:postgresql://gp-xxxxxx.gpdb.cn-chengdu.rds.aliyuncs.com:5432/postgres`.
tablename	Yes	The name of the AnalyticDB for PostgreSQL table.
username	Yes	The database account of the AnalyticDB for PostgreSQL instance.
password	Yes	The password of the database account for the AnalyticDB for PostgreSQL instance.
maxretrytimes	No	The maximum number of retries after an SQL execution failure. Default value: 3.
batchsize	No	The maximum number of data records to write in a single batch. Default value: 50000.
exceptionmode	No	The error handling policy when an exception occurs during data writing. Valid values: ignore: ignores the data that causes an exception. This is the default value. strict: triggers a failover and reports an error when a data writing exception occurs.
conflictmode	No	The policy for handling primary key or unique index conflicts. Valid values: ignore: ignores primary key conflicts and retains the existing data. strict: triggers a failover and reports an error when a primary key conflict occurs. update: updates the data when a primary key conflict occurs. upsert: writes data by using the UPSERT method when a primary key conflict occurs. This is the default value. AnalyticDB for PostgreSQL implements UPSERT by using INSERT ON CONFLICT and COPY ON CONFLICT. If the destination table is a partitioned table, the minor kernel version must be V6.3.6.1 or later. For more information about how to upgrade the minor kernel version, see Upgrade the engine version.
targetschema	No	The schema of the AnalyticDB for PostgreSQL database. Default value: public.
writemode	No	The data write mode. Valid values: 0: writes data by using BATCH INSERT. 1: writes data by using the COPY API. This is the default value. 2: writes data by using BATCH UPSERT.
verbose	No	Specifies whether to output connector runtime logs. Valid values: 0: does not output runtime logs. This is the default value. 1: outputs runtime logs.
retrywaittime	No	The interval between retry attempts when an exception occurs. Unit: milliseconds. Default value: 100.
batchwritetimeoutms	No	The maximum batch accumulation time for batch writing. When this time is exceeded, the accumulated batch is written. Unit: milliseconds. Default value: 50000.
connectionmaxactive	No	The connection pool parameter. This parameter specifies the maximum number of concurrent connections in the connection pool of a single Task Manager. Default value: 5.
casesensitive	No	Specifies whether column names and table names are case-sensitive. Valid values: 0: case-insensitive. This is the default value. 1: case-sensitive.

Note

Parameter and type mappings are supported. For more information, see the connector documentation for AnalyticDB for PostgreSQL (ADB PG).

Start the job.
1. At the top of the job development page, click Deploy. In the dialog box that appears, click OK.
  
  Note
  Session clusters are suitable for development and testing in non-production environments. You can use Session clusters to debug jobs, improve Job Manager (JM) resource utilization, and accelerate job startup. However, we do not recommend that you submit jobs to Session clusters due to business stability concerns. For more information, see Debug a job.
2. On the Deployments page, click Resume in the Actions column of the target job.
3. Click Resume.

Verify the results

Connect to the AnalyticDB for PostgreSQL database. For more information, see Connect to an instance by using a client.
Execute the following statement to query the test_adbpg_table table.
```
SELECT * FROM test_adbpg_table;
```
Data is written to AnalyticDB for PostgreSQL as expected. The following figure shows a sample return.

The query result contains three columns: b1 (int4), b2 (int4), and b3 (text). Multiple rows of data are returned, which confirms that data has been synchronized to the destination table.