AnalyticDB:Realtime\ Compute\ for\ Apache\ Flink\ を使用したベクトルデータの統合

前提条件

開始する前に、以下が準備できていることを確認してください。

• AnalyticDB for PostgreSQL インスタンス。セットアップ手順については、「インスタンスの作成」をご参照ください。

• AnalyticDB for PostgreSQL インスタンスと同じ VPC 内にある Flink 完全管理ワークスペース。セットアップ手順については、「Flink 完全管理のアクティベーション」をご参照ください。

  • オープンソースの自己管理 Flink を使用する場合は、$FLINK_HOME/lib ディレクトリに flink-adbpg-connector をインストールしてください。

  • Flink 完全管理を使用する場合、追加のインストールは不要です。

• AnalyticDB for PostgreSQL データベースに FastANN 拡張がインストールされていること。psql クライアントで \dx fastann を実行して確認します。出力が返されない場合は、チケットを起票してインストールを依頼してください。

• AnalyticDB for PostgreSQL インスタンスと同じ VPC 内にある ApsaraMQ for Kafka インスタンス。 詳細については、「インターネットおよび VPC 接続インスタンスを購入してデプロイする」をご参照ください。

• Flink ワークスペースと ApsaraMQ for Kafka インスタンスの CIDR ブロックが、AnalyticDB for PostgreSQL インスタンスの IP アドレスホワイトリストに追加されていること。手順については、「IP アドレスホワイトリストの設定」をご参照ください。

テストデータ

AnalyticDB for PostgreSQL では、手順に沿って操作するためのサンプルデータファイル、vector_sample_data.csv を提供しています。このファイルは以下のスキーマを使用します：

Linux でファイルをダウンロードするには：

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

仕組み

1. AnalyticDB for PostgreSQL に、構造化インデックスとベクトルインデックスを持つ送信先テーブルを作成します。

2. ベクトルテストデータを ApsaraMQ for Kafka トピックに発行します。

3. Kafka ソースと AnalyticDB for PostgreSQL シンクの両方に対して Flink マッピングテーブルを作成し、INSERT 文を実行してストリーミングインポートを開始します。

ステップ 1：送信先テーブルとインデックスの作成

1. psql クライアントを使用して AnalyticDB for PostgreSQL データベースに接続します。手順については、「psql」セクションのクライアント接続に関するトピックをご参照ください。

2. テストデータベースを作成して切り替えます。

   CREATE DATABASE adbpg_test;
   \c adbpg_test

3. 送信先テーブルを作成します。

   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. 構造化インデックスとベクトルインデックスを作成します。

   -- ベクトル列のストレージフォーマットを PLAIN に設定
   ALTER TABLE vector_test.car_info ALTER COLUMN feature SET STORAGE PLAIN;
   
   -- フィルタリング可能な列に構造化インデックスを作成
   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);
   
   -- 近似最近傍 (ANN) 検索を使用してベクトルインデックスを作成
   CREATE INDEX ON vector_test.car_info USING ann(feature)
   WITH (dim='10', pq_enable='0');

ステップ 2：ベクトルテストデータを Kafka トピックに発行

1. vector_ingest という名前の Kafka トピックを作成します。

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

2. テストデータをトピックに発行します。

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

<your_broker_list> を、ApsaraMQ for Kafka インスタンスのエンドポイントに置き換えてください。このエンドポイントは、ApsaraMQ for Kafka コンソールの[インスタンスの詳細]ページにある[エンドポイント情報]セクションで確認できます。

ステップ 3：マッピングテーブルの作成とインポートの開始

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;

次のステップ

• Realtime Compute for Apache Flink を使用して AnalyticDB for PostgreSQL にデータを書き込む — コネクタ パラメーターの完全なリファレンス

• IP アドレスホワイトリストの設定 — インスタンスのネットワークアクセスを管理