AnalyticDB:Flink için Realtime Compute ile vektör verilerini entegre etme

Ön koşullar

• Bir AnalyticDB for PostgreSQL örneği oluşturun. Daha fazla bilgi için bkz. Örnek oluşturma.

• Tam yönetilen bir Flink çalışma alanı oluşturun ve bu alanın AnalyticDB for PostgreSQL örneğiyle aynı VPC içinde olduğundan emin olun. Daha fazla bilgi için bkz. Apache Flink için Realtime Compute'u etkinleştirme.

  • Kendi kendinize yönetilen, açık kaynaklı bir Flink kümesi kullanıyorsanız flink-adbpg-connector paketinin $FLINK_HOME/lib dizininde yüklü olduğundan emin olun.

  • tam yönetilen hizmeti kullanıyorsanız herhangi bir işlem yapmanız gerekmez.

• FastANN vektör getirme uzantısını AnalyticDB for PostgreSQL veritabanınıza yükleyin.

  Uzantının yüklü olup olmadığını kontrol etmek için psql istemcisinde \dx fastann komutunu çalıştırabilirsiniz.

  • FastANN uzantısına ilişkin bilgiler döndürülüyorsa uzantı yüklenmiştir.

  • Herhangi bir bilgi döndürülmüyorsa kurulum için teknik destekle iletişime geçmek üzere Destek talebi gönderin.

• Bir Message Queue for Apache Kafka örneği satın alın ve AnalyticDB for PostgreSQL örneğiyle aynı VPC içinde konuşlandırın. Daha fazla bilgi için bkz. Örnek satın alma ve konuşlandırma.

• Flink çalışma alanı ve Kafka örneğine ait CIDR bloklarını AnalyticDB for PostgreSQL örneğinin IP adresi izin listesine ekleyin. Daha fazla bilgi için bkz. IP adresi izin listesi yapılandırma.

Örnek veriler

AnalyticDB for PostgreSQL test amaçlı örnek veriler sağlar. Verileri indirmek için vector_sample_data.csv bağlantısını tıklayın.

Aşağıdaki tabloda örnek veri şeması açıklanmaktadır.

Yordam

1. Yapısal ve vektör dizinleri oluşturun.

2. Vektör örnek verilerini bir Kafka konusuna yazın.

3. Eşleme tablosu oluşturun ve verileri içe aktarın.

Yapısal ve vektör dizinleri oluşturma

1. AnalyticDB for PostgreSQL veritabanınıza bağlanın. Aşağıdaki adımlarda psql istemcisi kullanılmaktadır. Daha fazla bilgi için bkz. psql kullanarak veritabanına bağlanma.

2. Test veritabanı oluşturmak ve bu veritabanına geçiş yapmak için aşağıdaki deyimleri çalıştırın:

   CREATE DATABASE adbpg_test;
   \c adbpg_test

3. Hedef tabloyu oluşturmak için aşağıdaki deyimleri çalıştırın:

   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. Yapısal ve vektör dizinleri oluşturmak için aşağıdaki deyimleri çalıştırın:

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

Örnek vektör verilerini Kafka'ya yazma

1. Kafka konusu oluşturmak için aşağıdaki komutu çalıştırın:

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

2. Vektör örnek verilerini Kafka konusuna yazmak için aşağıdaki komutu çalıştırın:

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

<your_broker_list>: Örnek uç noktası. Uç noktayı Message Queue for Apache Kafka konsolundaki Instance Details sayfasının Access Point Information bölümünden elde edebilirsiniz.

Eşleme tablosu oluşturma ve veri içe aktarma

1. Flink işi oluşturun.

   1. Realtime Compute for Apache Flink konsoluna giriş yapın. Fully Managed Flink sekmesinde hedef çalışma alanını bulun ve Actions sütunundaki Console öğesini tıklayın.

   2. Sol gezinme bölmesinde SQL Development seçeneğini tıklayın. New öğesini tıklayın, Blank stream draft seçeneğini belirleyin ve ardından Next düğmesini tıklayın.

   3. New draft iletişim kutusunda taslak parametrelerini yapılandırın.

      Parametre	Açıklama	Örnek
      File Name	Taslağın adı. Not Taslak adı mevcut projede benzersiz olmalıdır.	adbpg-test
      Storage Location	Taslağın kod dosyasının depolandığı klasör. Mevcut bir klasörün yanındaki simgesini tıklayarak da alt klasör oluşturabilirsiniz.	Drafts
      Engine Version	Geçerli iş için Flink altyapı sürümü. Altyapı sürümleri, sürüm eşlemeleri ve yaşam döngüsü kilometre taşları hakkında daha fazla bilgi için bkz. Altyapı sürümleri.	vvr-6.0.6-flink-1.15

2. AnalyticDB for PostgreSQL için eşleme tablosu oluşturmak üzere aşağıdaki deyimi çalıştırın:

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

   Parametre açıklamaları için bkz. AnalyticDB for PostgreSQL'e veri yazma.

3. Kafka için eşleme tablosu oluşturmak üzere aşağıdaki deyimi çalıştırın:

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

   Aşağıdaki tabloda parametreler açıklanmaktadır.

   Parametre	Gerekli	Açıklama
   connector	Evet	Bağlayıcı adı. Değer kafka olmalıdır.
   properties.bootstrap.servers	Evet	Message Queue for Apache Kafka örneğinin uç noktası. Uç noktayı Message Queue for Apache Kafka konsolundaki Instance details sayfasının Endpoint information bölümünden elde edebilirsiniz.
   topic	Evet	Kafka konusunun adı.
   format	Evet	Kafka mesaj değerlerinin biçimi. Aşağıdaki biçimler desteklenir: csv json avro debezium-json canal-json maxwell-json avro-confluent raw
   csv.field-delimiter	Evet	CSV biçimi için alan sınırlayıcısı.
   scan.startup.mode	Evet	Kafka tüketicisinin veri okumaya başlayacağı ofseti belirtir. Geçerli değerler: earliest-offset: En eski kullanılabilir ofsetten itibaren okumaya başlar. latest-offset: En son ofsetten itibaren okumaya başlar.

4. İçe aktarma işi oluşturmak için aşağıdaki deyimi çalıştırın:

   INSERT INTO vector_ingest SELECT * FROM vector_kafka;