Integrate vector data with Realtime Compute for Flink

Prerequisites

• Create an AnalyticDB for PostgreSQL instance. For more information, see Create an instance.

• A fully managed Flink workspace is created and is in the same VPC as the AnalyticDB for PostgreSQL instance. For more information, see Activate Realtime Compute for Apache Flink.

  • If you use a self-managed, open-source Flink cluster, make sure that the flink-adbpg-connector is installed in the $FLINK_HOME/lib directory.

  • If you use the fully managed service, no action is required.

• The FastANN vector retrieval extension is installed in your AnalyticDB for PostgreSQL database.

  You can run the \dx fastann command on a psql client to check whether the extension is installed.

  • If information about the FastANN extension is returned, the extension is installed.

  • If no information is returned, Submit a ticket to contact technical support for installation.

• A Message Queue for Apache Kafka instance has been purchased and deployed in the same VPC as the AnalyticDB for PostgreSQL instance. For more information, see Purchase and deploy an instance.

• The CIDR blocks for the Flink workspace and the Kafka instance have been added to the IP address whitelist of the AnalyticDB for PostgreSQL instance. For more information, see Configure an IP address whitelist.

Sample data

AnalyticDB for PostgreSQL provides sample data for testing. To download the data, click vector_sample_data.csv.

The following table describes the sample data schema.

Procedure

1. Create structured and vector indexes.

2. Write the vector sample data to a Kafka topic.

3. Create a mapping table and import data.

Create structured and vector indexes

1. Connect to your AnalyticDB for PostgreSQL database. The following steps use a psql client. For more information, see Connect to a database using psql.

2. Run the following statements to create and switch to a test database:

   CREATE DATABASE adbpg_test;
   \c adbpg_test

3. Run the following statements to create a destination table:

   CREATE SCHEMA IF NOT EXISTS vector_test;
   CREATE TABLE IF NOT EXISTS vector_test.car_info
   (
     id bigint NOT NULL,
     market_time timestamp,
     color varchar(10),
     price int,
     feature float4[],
     PRIMARY KEY(id)
   ) DISTRIBUTED BY(id);

4. Run the following statements to create structured and vector indexes:

   -- Change the storage format of the vector column to PLAIN.
   ALTER TABLE vector_test.car_info ALTER COLUMN feature SET STORAGE PLAIN;
   
   -- Create structured indexes.
   CREATE INDEX ON vector_test.car_info(market_time);
   CREATE INDEX ON vector_test.car_info(color);
   CREATE INDEX ON vector_test.car_info(price);
   
   -- Create a vector index.
   CREATE INDEX ON vector_test.car_info USING ann(feature) 
   WITH (dim='10', pq_enable='0');

Write sample vector data to Kafka

1. Run the following command to create a Kafka topic:

   bin/kafka-topics.sh --create --topic vector_ingest --partitions 1 \
   --bootstrap-server <your_broker_list>

2. Run the following command to write the vector sample data to the Kafka topic:

   bin/kafka-console-producer.sh \
   --bootstrap-server <your_broker_list> \
   --topic vector_ingest < ../vector_sample_data.csv

<your_broker_list>: The instance endpoint. You can obtain the endpoint from the Access Point Information section of the Instance Details page in the Message Queue for Apache Kafka console.

Create a mapping table and import data

1. Create a Flink job.

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

   2. In the left-side navigation pane, click SQL Development. Click Create, select Blank stream draft, and then click Next.

   3. In the New draft dialog box, configure the draft parameters.

      Parameter	Description	Example
      File Name	The name of the draft. Note The draft name must be unique in the current project.	adbpg-test
      Storage Location	The folder where the draft's code file is stored. You can also click the icon next to an existing folder to create a subfolder.	Drafts
      Engine Version	The Flink engine version for the current job. For more information about engine versions, version mappings, and lifecycle milestones, see Engine versions.	vvr-6.0.6-flink-1.15

2. Run the following statement to create a mapping table for AnalyticDB for PostgreSQL:

   CREATE TABLE vector_ingest (
     id INT,
     market_time TIMESTAMP,
     color VARCHAR(10),
     price int,
     feature VARCHAR
   )WITH (
      'connector' = 'adbpg-nightly-1.13',
      'url' = 'jdbc:postgresql://<your_instance_url>:5432/adbpg_test',
      'tablename' = 'car_info',
      'username' = '<your_username>',
      'password' = '<your_password>',
      'targetschema' = 'vector_test',
      'maxretrytimes' = '2',
      'batchsize' = '3000',
      'batchwritetimeoutms' = '10000',
      'connectionmaxactive' = '20',
      'conflictmode' = 'ignore',
      'exceptionmode' = 'ignore',
      'casesensitive' = '0',
      'writemode' = '1',
      'retrywaittime' = '200'
   );

   For parameter descriptions, see Write data to AnalyticDB for PostgreSQL.

3. Run the following statement to create a mapping table for Kafka:

   CREATE TABLE vector_kafka (
     id INT,
     market_time TIMESTAMP,
     color VARCHAR(10),
     price int,
     feature string
   ) 
   WITH (
       'connector' = 'kafka',
       'properties.bootstrap.servers' = '<your_broker_list>',
       'topic' = 'vector_ingest',
       'format' = 'csv',
       'csv.field-delimiter' = '\t',
       'scan.startup.mode' = 'earliest-offset'
   );

   The following table describes the parameters.

   Parameter	Required	Description
   connector	Yes	The connector name. The value must be kafka.
   properties.bootstrap.servers	Yes	The endpoint of the Message Queue for Apache Kafka instance. You can obtain the endpoint from the Endpoint information section of the Instance details page in the Message Queue for Apache Kafka console.
   topic	Yes	The name of the Kafka topic.
   format	Yes	The format of Kafka message values. The following formats are supported: csv json avro debezium-json canal-json maxwell-json avro-confluent raw
   csv.field-delimiter	Yes	The field delimiter for the CSV format.
   scan.startup.mode	Yes	Specifies the offset from which the Kafka consumer starts reading data. Valid values: earliest-offset: Starts reading from the earliest available offset. latest-offset: Starts reading from the latest offset.

4. Run the following statement to create an import job:

   INSERT INTO vector_ingest SELECT * FROM vector_kafka;