AnalyticDB:Integrate vector data using Realtime Compute for Apache Flink

Prerequisites

Before you begin, ensure that you have:

• An AnalyticDB for PostgreSQL instance. For setup instructions, see Create an instance.

• A fully managed Flink workspace in the same virtual private cloud (VPC) as the AnalyticDB for PostgreSQL instance. For setup instructions, see Activate fully managed Flink.

  • If you use open source self-managed Flink, install flink-adbpg-connector in the $FLINK_HOME/lib directory.

  • If you use fully managed Flink, no additional installation is required.

• The FastANN extension installed in the AnalyticDB for PostgreSQL database. Run \dx fastann on the psql client to check. If no output is returned, Submit a ticket to have it installed.

• An ApsaraMQ for Kafka instance in the same VPC as the AnalyticDB for PostgreSQL instance. For setup instructions, see Purchase and deploy an Internet- and VPC-connected instance.

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

Test data

AnalyticDB for PostgreSQL provides a sample data file, vector_sample_data.csv, to help you follow along. The file uses the following schema:

To download the file on Linux:

wget https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20230606/uzkx/vector_sample_data.csv

How it works

1. Create a destination table in AnalyticDB for PostgreSQL with structured indexes and a vector index.

2. Publish the vector test data to an ApsaraMQ for Kafka topic.

3. Create Flink mapping tables for both the Kafka source and the AnalyticDB for PostgreSQL sink, then run an INSERT statement to start the streaming import.

Step 1: Create the destination table and indexes

1. Connect to the AnalyticDB for PostgreSQL database using the psql client. For instructions, see the psql section of the client connection topic.

2. Create and switch to a test database:

   CREATE DATABASE adbpg_test;
   \c adbpg_test

3. Create the 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. Create structured indexes and a vector index:

   -- Set the storage format of the vector column to PLAIN
   ALTER TABLE vector_test.car_info ALTER COLUMN feature SET STORAGE PLAIN;
   
   -- Create structured indexes on filterable columns
   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 using approximate nearest neighbor (ANN) search
   CREATE INDEX ON vector_test.car_info USING ann(feature)
   WITH (dim='10', pq_enable='0');

Step 2: Publish vector test data to a Kafka topic

1. Create a Kafka topic named vector_ingest:

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

2. Publish the test data to the topic:

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

Replace <your_broker_list> with the endpoint of your ApsaraMQ for Kafka instance. Find the endpoint in the Endpoint Information section of the Instance Details page in the ApsaraMQ for Kafka console.

Step 3: Create mapping tables and start the import

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'
);

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'
);

INSERT INTO vector_ingest SELECT * FROM vector_kafka;

What's next

• Use Realtime Compute for Apache Flink to write data to AnalyticDB for PostgreSQL — full connector parameter reference

• Configure an IP address whitelist — manage network access for your instance