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

Step 1: Create the Kafka source table in Flink

On the Flink client, define a source table that reads from the Kafka topic:

CREATE TABLE KafkaSourceTable (
  `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 event_time, shard, and meta_offset columns capture Kafka message metadata. shard maps from the Kafka partition metadata field, and meta_offset maps from the offset metadata field — both renamed to avoid conflicts with AnalyticDB for PostgreSQL reserved keywords. For all available metadata fields, see Available Metadata.

Kafka connector parameters

For the full parameter reference, see Connector options.

Step 2: Create the destination table in AnalyticDB for PostgreSQL

Connect to the AnalyticDB for PostgreSQL instance and create the destination table. Use a schema that matches the Flink source table, replacing reserved keywords with alternative column names:

CREATE TABLE ADBPGTargetTable (
  user_id     BIGINT PRIMARY KEY,
  item_id     BIGINT,
  behavior    VARCHAR,
  event_time  TIMESTAMP,
  shard       BIGINT,       -- replaces 'partition' (reserved keyword)
  meta_offset BIGINT        -- replaces 'offset' (reserved keyword)
);

Step 3: Create the AnalyticDB for PostgreSQL sink table in Flink

On the Flink client, define a sink table that maps to the AnalyticDB for PostgreSQL destination table:

CREATE TABLE ADBPGTargetTable (
  `user_id`     BIGINT PRIMARY KEY,
  `item_id`     BIGINT,
  `behavior`    STRING,
  `event_time`  TIMESTAMP(3),
  `shard`       BIGINT,
  `meta_offset` BIGINT
) WITH (
  'connector'           = 'adbpg-nightly-1.13',
  'url'                 = 'jdbc:postgresql://gp-bp15s3b9kn00j****-master.gpdb.rds.aliyuncs.com:5432/postgres',
  'tablename'           = 'ADBPGTargetTable',
  'username'            = 'user01',
  'password'            = 'Password01',
  'targetschema'        = 'public',
  'writemode'           = '1',
  'batchsize'           = '50000',
  'conflictmode'        = 'ignore',
  'exceptionmode'       = 'ignore',
  'maxretrytimes'       = '2',
  'retrywaittime'       = '200',
  'connectionmaxactive' = '5',
  'casesensitive'       = '0',
  'usecopy'             = '0'
);

AnalyticDB for PostgreSQL connector parameters

Step 4: Run the synchronization job

On the Flink client, run the following statement to start continuous data ingestion from Kafka into AnalyticDB for PostgreSQL:

INSERT INTO ADBPGTargetTable SELECT * FROM KafkaSourceTable;

The Flink client console shows the job execution status.