AnalyticDB:Integración de datos vectoriales con Realtime Compute for Flink

Requisitos previos

• Se creó una instancia de AnalyticDB for PostgreSQL. Para obtener más información, consulte Create an instance.

• Se creó un espacio de trabajo de Flink totalmente gestionado y se encuentra en la misma VPC que la instancia de AnalyticDB for PostgreSQL. Para obtener más información, consulte Activate Realtime Compute for Apache Flink.

  • Si utiliza un clúster de Flink de código abierto autogestionado, asegúrese de que flink-adbpg-connector esté instalado en el directorio $FLINK_HOME/lib.

  • Si utiliza el servicio totalmente gestionado, no se requiere ninguna acción.

• La extensión de recuperación vectorial FastANN está instalada en su base de datos de AnalyticDB for PostgreSQL.

  Puede ejecutar el comando \dx fastann en un cliente psql para comprobar si la extensión está instalada.

  • Si se devuelve información sobre la extensión FastANN, significa que la extensión está instalada.

  • Si no se devuelve ninguna información, Submit a ticket para ponerse en contacto con el soporte técnico y solicitar la instalación.

• Se adquirió e implementó una instancia de Message Queue for Apache Kafka en la misma VPC que la instancia de AnalyticDB for PostgreSQL. Para obtener más información, consulte Purchase and deploy an instance.

• Los bloques CIDR del espacio de trabajo de Flink y de la instancia de Kafka se agregaron a la lista blanca de direcciones IP de la instancia de AnalyticDB for PostgreSQL. Para obtener más información, consulte Configure an IP address whitelist.

Datos de ejemplo

AnalyticDB for PostgreSQL proporciona datos de ejemplo para pruebas. Para descargar los datos, haga clic en vector_sample_data.csv.

La siguiente tabla describe el esquema de los datos de ejemplo.

Procedimiento

1. Create structured and vector indexes.

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

3. Create a mapping table and import data.

Creación de índices estructurados y vectoriales

1. Conéctese a su base de datos de AnalyticDB for PostgreSQL. Los siguientes pasos utilizan un cliente psql. Para obtener más información, consulte Connect to a database using psql.

2. Ejecute las siguientes sentencias para crear una base de datos de prueba y cambiar a ella:

   CREATE DATABASE adbpg_test;
   \c adbpg_test

3. Ejecute las siguientes sentencias para crear una tabla de destino:

   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. Ejecute las siguientes sentencias para crear índices estructurados y vectoriales:

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

Escritura de datos vectoriales de ejemplo en Kafka

1. Ejecute el siguiente comando para crear un topic de Kafka:

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

2. Ejecute el siguiente comando para escribir los datos vectoriales de ejemplo en el topic de Kafka:

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

<your_broker_list>: El endpoint de la instancia. Puede obtener el endpoint en la sección Access Point Information de la página Instance Details en la consola de Message Queue for Apache Kafka.

Creación de una tabla de asignación e importación de datos

1. Cree un trabajo de Flink.

   1. Inicie sesión en la consola de Realtime Compute for Apache Flink. En la pestaña Fully Managed Flink, localice el espacio de trabajo deseado y haga clic en Console en la columna Actions.

   2. En el panel de navegación izquierdo, haga clic en SQL Development. Haga clic en New, seleccione Blank stream draft y luego haga clic en Next.

   3. En el cuadro de diálogo New draft, configure los parámetros del borrador.

      Parámetro	Descripción	Ejemplo
      File Name	El nombre del borrador. Nota El nombre del borrador debe ser único dentro del proyecto actual.	adbpg-test
      Storage Location	La carpeta donde se almacena el archivo de código del borrador. También puede hacer clic en el icono junto a una carpeta existente para crear una subcarpeta.	Drafts
      Engine Version	La versión del motor de Flink para el trabajo actual. Para obtener más información sobre las versiones del motor, las asignaciones de versiones y los hitos del ciclo de vida, consulte Engine versions.	vvr-6.0.6-flink-1.15

2. Ejecute la siguiente sentencia para crear una tabla de asignación para 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'
   );

   Para consultar la descripción de los parámetros, véase Write data to AnalyticDB for PostgreSQL.

3. Ejecute la siguiente sentencia para crear una tabla de asignación para 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'
   );

   La siguiente tabla describe los parámetros.

   Parámetro	Obligatorio	Descripción
   connector	Sí	El nombre del conector. El valor debe ser kafka.
   properties.bootstrap.servers	Sí	El endpoint de la instancia de Message Queue for Apache Kafka. Puede obtener el endpoint en la sección de información del endpoint de la página de detalles de la instancia en la consola de Message Queue for Apache Kafka.
   topic	Sí	El nombre del topic de Kafka.
   format	Sí	El formato de los valores de los mensajes de Kafka. Se admiten los siguientes formatos: csv json avro debezium-json canal-json maxwell-json avro-confluent raw
   csv.field-delimiter	Sí	El delimitador de campos para el formato CSV.
   scan.startup.mode	Sí	Especifica el desplazamiento desde el cual el consumidor de Kafka comienza a leer datos. Valores válidos: earliest-offset: Comienza a leer desde el desplazamiento disponible más antiguo. latest-offset: Comienza a leer desde el desplazamiento más reciente.

4. Ejecute la siguiente sentencia para crear un trabajo de importación:

   INSERT INTO vector_ingest SELECT * FROM vector_kafka;