AnalyticDB:Use Apache Flink to synchronize data from Kafka to AnalyticDB for PostgreSQL

Usage notes

The use of Alibaba Cloud Realtime Compute for Apache Flink is similar to that of Apache Flink. For more information, see What is Alibaba Cloud Realtime Compute for Apache Flink?

Procedure

1. On the Flink client, create a table that is used to read data from a Kafka table.

   CREATE TABLE KafkaTable (
     `user_id` BIGINT,
     `item_id` BIGINT,
     `behavior` STRING,
     `event_time` TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,
     `shard` BIGINT METADATA FROM 'partition' VIRTUAL,
     `meta_offset` BIGINT METADATA FROM 'offset' VIRTUAL
   
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'user_behavior',
     'properties.bootstrap.servers' = 'localhost:9092',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'earliest-offset',
     'value.format' = 'debezium-json'
   );

   The following table describes the Kafka connector parameters.

   Parameter	Required	Description
   connector	Yes	The name of the connector. The value is automatically set to kafka.
   topic	Yes	The name of the Kafka topic.
   properties.bootstrap.servers	Yes	The endpoint and port number that are used to connect to the Kafka client.
   properties.group.id	Yes	The ID of the Kafka consumer group.
   scan.startup.mode	No	The starting consumer offset. For more information, see Start Reading Position.
   value.format	Yes	The format that is used to serialize and deserialize the Kafka message. For more information about the formats and configurations, see Formats.

   For more information about Kafka connector parameters, see Connector Options.

   Each Kafka message record consists of useful metadata such as the timestamp, offset, and partition information. The event_time, meta_offset, and shard columns in the test table record the useful information of Kafka messages. For more information about available metadata, see Available Metadata.

2. In the AnalyticDB for PostgreSQL instance, create a destination table.

   CREATE TABLE ADBPGTargetTable (
     user_id BIGINT primary key,
     item_id BIGINT,
     behavior VARCHAR,
     event_time TIMESTAMP,
     shard BIGINT,            -- The partition column of the Kafka table is replaced with the shard column because partition is a reserved keyword in AnalyticDB for PostgreSQL. 
     meta_offset BIGINT       -- The offset column of the Kafka table is replaced with the meta_offset column because offset is a reserved keyword in AnalyticDB for PostgreSQL. 
   );

3. On the Flink client, create a table that is used to synchronize data to the AnalyticDB for PostgreSQL instance. We recommend that you use the same schema as that of the source table created in Step 1.

   CREATE TABLE ADBPGTargetTable (
     `user_id` BIGINT primary key,
     `item_id` BIGINT,
     `behavior` STRING,
     `event_time` TIMESTAMP(3),
     `shard` BIGINT,            -- The partition column of the Kafka table is replaced with the shard column because partition is a reserved keyword in AnalyticDB for PostgreSQL. 
     `meta_offset` BIGINT       -- The offset column of the Kafka table is replaced with the meta_offset column because offset is a reserved keyword in AnalyticDB for PostgreSQL. 
   ) WITH (
      'connector' = 'adbpg-nightly-1.13',
      'password' = 'Password01',
      'tablename' = 'ADBPGTargetTable',
      'username' = 'user01',
      'url' = 'jdbc:postgresql://gp-bp15s3b9kn00j****-master.gpdb.rds.aliyuncs.com:5432/postgres',
      'maxretrytimes' = '2',
      'batchsize' = '50000',
      'connectionmaxactive' = '5',
      'conflictmode' = 'ignore',
      'usecopy' = '0',
      'targetschema' = 'public',
      'exceptionmode' = 'ignore',
      'casesensitive' = '0',
      'writemode' = '1',
      'retrywaittime' = '200'
   );

   The following table describes the AnalyticDB for PostgreSQL connector parameters.

   Parameter	Required	Description
   connector	Yes	The name of the connector. The value is in the adbpg-nightly-Version number format. For example, if AnalyticDB for PostgreSQL connector 1.13 is used, the connector name is adbpg-nightly-1.13.
   url	Yes	The JDBC URL that is used to connect to the AnalyticDB for PostgreSQL instance. It is in the jdbc:postgresql://<Endpoint:port>/<Database name> format. Example: jdbc:postgresql://gp-bp15s3b9kn00j****-master.gpdb.rds.aliyuncs.com:5432/postgres.
   tablename	Yes	The name of the table that is created in the AnalyticDB for PostgreSQL instance.
   username	Yes	The name of the database account that is used to connect to the AnalyticDB for PostgreSQL instance.
   password	Yes	The password of the database account that is used to connect to the AnalyticDB for PostgreSQL instance.
   maxretrytimes	No	The maximum number of retries that are allowed after a statement fails to be executed. Default value: 3.
   batchsize	No	The maximum number of rows that can be written at a time. Default value: 50000.
   exceptionmode	No	The policy that is used to handle exceptions during data writing. Default value: ignore. Valid values: ignore: If an exception occurs during data writing, the system ignores the error data. strict: If an exception occurs during data writing, the system performs a failover and reports an error.
   conflictmode	No	The policy that is used to handle primary key or unique index conflicts. Default value: upsert. Valid values: ignore: If a primary key conflict occurs, the system ignores the primary key conflict and retains the existing data. strict: If a primary key conflict occurs, the system performs a failover and reports an error. update: If a primary key conflict occurs, the system updates data. upsert: If a primary key conflict occurs, the system performs UPSERT operations to write data. AnalyticDB for PostgreSQL uses INSERT ON CONFLICT and COPY ON CONFLICT to implement UPSERT operations. If the destination table is a partitioned table, the AnalyticDB for PostgreSQL instance must run minor version 6.3.6.1 or later. For more information about how to update the minor engine version of your instance, see Update the minor engine version.
   targetschema	No	The schema of the AnalyticDB for PostgreSQL instance. Default value: public.
   writemode	No	The method that is used to write data. Default value: 1. Valid values: 0: The BATCH INSERT statement is used to write data. 1: The COPY API is used to write data. 2: The BATCH UPSERT statement is used to write data.
   verbose	No	Specifies whether to display connector logs. Default value: 0. Valid values: 0: does not display connector logs. 1: displays connector logs.
   retrywaittime	No	The retry interval after an exception occurs. Default value: 100. Unit: milliseconds.
   batchwritetimeoutms	No	The timeout period for batch data writing. After this period ends, data is batch written. Default value: 50000. Unit: milliseconds.
   connectionmaxactive	No	The maximum number of active connections that can be allocated in a connection pool for a single task manager. Default value: 5.
   casesensitive	No	Specifies whether column and table names are case-sensitive. Default value: 0. Valid values: 0: case-insensitive 1: case-sensitive

4. On the Flink client, execute the INSERT INTO statement to synchronize data from the Kafka table to the AnalyticDB for PostgreSQL table.

   INSERT INTO ADBPGTargetTable SELECT * FROM KafkaSourceTable;

The execution status is displayed in the console of the Flink client, as shown in the following figure.