Synchronize data from Kafka to AnalyticDB for PostgreSQL

Usage notes

The procedure for Alibaba Cloud Realtime Compute for Apache Flink is similar to that for Apache Flink, with some operational differences. For more information, see Alibaba Cloud Realtime Compute for Apache Flink documentation.

Procedure

1. Create a table in Flink to read data from a Kafka topic.

   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. Set this to kafka.
   topic	Yes	The name of the Kafka topic.
   properties.bootstrap.servers	Yes	The addresses and ports of the Kafka brokers.
   properties.group.id	Yes	The consumer group ID.
   scan.startup.mode	No	start reading positionThe startup mode for the Kafka consumer. For more information, see .
   value.format	Yes	FormatsThe format used to serialize and deserialize the Kafka message body. For more information about formats and their configurations, see .

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

   Each Kafka message (Record) contains metadata, such as timestamp, offset, and partition, that can be useful in your applications. The event_time, meta_offset, and shard columns in the test table are examples of useful information retrieved from Kafka messages. For more information about the available metadata, see Available Metadata.

2. Create a destination table in AnalyticDB for PostgreSQL.

   CREATE TABLE ADBPGTargetTable (
     user_id BIGINT primary key,
     item_id BIGINT,
     behavior VARCHAR,
     event_time TIMESTAMP,
     shard BIGINT,            -- 'partition' is a reserved keyword in AnalyticDB for PostgreSQL. The original partition column from the Kafka table is renamed to shard.
     meta_offset BIGINT       -- 'offset' is a reserved keyword in AnalyticDB for PostgreSQL. The original offset column from the Kafka table is renamed to meta_offset.
   );

3. Create a table in Flink to write data to AnalyticDB for PostgreSQL. We recommend that this table has the same schema as 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,            -- 'partition' is a reserved keyword in AnalyticDB for PostgreSQL. The original partition column from the Kafka table is renamed to shard.
     `meta_offset` BIGINT       -- 'offset' is a reserved keyword in AnalyticDB for PostgreSQL. The original offset column from the Kafka table is renamed to meta_offset.
   ) 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 format is adbpg-nightly-<version>. For example, if you use version 1.13 of the AnalyticDB for PostgreSQL connector, the name is adbpg-nightly-1.13.
   url	Yes	The JDBC connection string for AnalyticDB for PostgreSQL. The format is jdbc:postgresql://<host:port>/<database_name>. Example: jdbc:postgresql://gp-bp15s3b9kn00j****-master.gpdb.rds.aliyuncs.com:5432/postgres.
   tablename	Yes	The name of the table in AnalyticDB for PostgreSQL.
   username	Yes	The database account.
   password	Yes	The password for the database account.
   maxretrytimes	No	The number of retries after an SQL execution fails. Default value: 3.
   batchsize	No	The maximum number of records for a batch write. Default value: 50000.
   exceptionmode	No	The policy for handling exceptions during write operations. Supported policies: ignore (default): Ignores the data that causes the exception. strict: Triggers a failover and reports an error.
   conflictmode	No	The policy for handling a primary key conflict or unique index conflict. Supported policies: ignore: Ignores the conflict and keeps the existing data. strict: Triggers a failover and reports an error. update: Updates the existing record with the new data. upsert (default): Uses an UPSERT operation to write data. AnalyticDB for PostgreSQL implements UPSERT operations by using INSERT ON CONFLICT and COPY ON CONFLICT. If the destination table is a partitioned table, the instance must have a minor engine version of V6.3.6.1 or later. For more information about how to perform a version upgrade, see Upgrade the Minor Engine Version.
   targetschema	No	The schema in AnalyticDB for PostgreSQL. Default value: public.
   writemode	No	The write mode. Valid values: 0: Uses BATCH INSERT to write data. 1 (default): Uses the COPY API to write data. 2: Uses BATCH UPSERT to write data.
   verbose	No	Specifies whether to output connector logs. Valid values: 0 (default): Does not output logs. 1: Outputs logs.
   retrywaittime	No	The interval between retries after an exception occurs. Unit: milliseconds (ms). Default value: 100.
   batchwritetimeoutms	No	The maximum time to buffer data for a batch write. The write is triggered when this time is exceeded. Unit: milliseconds (ms). Default value: 50000.
   connectionmaxactive	No	The maximum number of active connections in the connection pool for a single TaskManager. Default value: 5.
   casesensitive	No	Specifies whether column and table names are case-sensitive. Valid values: 0 (default): Case-insensitive. 1: Case-sensitive.

4. Run the INSERT INTO statement in Flink to synchronize data from the Kafka source table to AnalyticDB for PostgreSQL.

   INSERT INTO ADBPGTargetTable SELECT * FROM KafkaSourceTable;