AnalyticDB:Realtime Compute for Apache Flink とのベクトルデータの統合

前提条件

• AnalyticDB for PostgreSQL インスタンスが作成されます。詳細については、「インスタンスを作成する」をご参照ください。

• Flink 完全管理ワークスペースが作成され、AnalyticDB for PostgreSQL インスタンスと同じ VPC 内にある必要があります。詳細については、「Realtime Compute for Apache Flink のアクティベーション」をご参照ください。

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

  • フルマネージドサービスを使用する場合、操作は不要です。

• FastANN ベクトル検索拡張が、ご利用の AnalyticDB for PostgreSQL データベースにインストールされている必要があります。

  psql クライアントで `\dx fastann` コマンドを実行して、拡張がインストールされているかどうかを確認できます。

  • FastANN 拡張に関する情報が返された場合、拡張はインストールされています。

  • 情報が返されない場合は、チケットを起票してテクニカルサポートに連絡し、インストールを依頼してください。

• Message Queue for Apache Kafka インスタンスが購入・デプロイされ、AnalyticDB for PostgreSQL インスタンスと同じ VPC 内にある必要があります。詳細については、「インスタンスの購入とデプロイ」をご参照ください。

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

サンプルデータ

AnalyticDB for PostgreSQL では、テスト用のサンプルデータを提供しています。データをダウンロードするには、vector_sample_data.csv をクリックしてください。

次の表に、サンプルデータのスキーマを示します。

操作手順

1. 構造化インデックスとベクトルインデックスの作成

2. ベクトルサンプルデータの Kafka トピックへの書き込み

3. マッピングテーブルの作成とデータのインポート

構造化インデックスとベクトルインデックスの作成

1. ご利用の AnalyticDB for PostgreSQL データベースに接続します。以下の手順では psql クライアントを使用します。詳細については、「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);
   
   -- ベクターインデックスを作成します。
   CREATE INDEX ON vector_test.car_info USING ann(feature) 
   WITH (dim='10', pq_enable='0');

サンプルベクトルデータの Kafka への書き込み

1. 次のコマンドを実行して、Kafka トピックを作成します。

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

2. 次のコマンドを実行して、ベクトルサンプルデータを Kafka トピックに書き込みます。

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

<your_broker_list>: インスタンスのエンドポイントであり、Message Queue for Apache Kafka コンソールのインスタンス詳細 ページのアクセスポイント情報 セクションから取得できます。

マッピングテーブルの作成とデータのインポート

1. Flink ジョブを作成します。

   1. Realtime Compute for Apache Flink コンソールにログインします。完全マネージド Flink タブで、対象のワークスペースを探し、操作 列のコンソール をクリックします。

   2. 左側のナビゲーションウィンドウで、[SQL 開発] をクリックします。New をクリックし、[空白のストリームジョブの下書き] を選択して、次へ をクリックします。

   3. [新規ドラフト] ダイアログボックスで、ドラフトパラメーターを設定します。

      パラメーター	説明	例
      File Name	下書きの名前。 説明 下書き名は、現在のプロジェクト内で一意である必要があります。	adbpg-test
      Storage Location	下書きのコードファイルが保存されるフォルダ。 既存のフォルダの横にある アイコンをクリックして、サブフォルダを作成することもできます。	下書き
      エンジンバージョン	現在のジョブの Flink エンジンバージョン。エンジンバージョン、バージョンマッピング、ライフサイクルマイルストーンの詳細については、「エンジンバージョン」をご参照ください。	vvr-6.0.6-flink-1.15

2. 次の文を実行して、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'
   );

   パラメーターの説明については、「AnalyticDB for PostgreSQL へのデータの書き込み」をご参照ください。

3. 次の文を実行して、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'
   );

   次の表にパラメーターを示します。

   パラメーター	必須	説明
   connector	はい	コネクタ名。値は `kafka` である必要があります。
   properties.bootstrap.servers	はい	Message Queue for Apache Kafka インスタンスのエンドポイント。エンドポイントは、Message Queue for Apache Kafka コンソールの [インスタンス詳細] ページの [エンドポイント情報] セクションから取得できます。
   topic	はい	Kafka トピックの名前。
   format	はい	Kafka メッセージ値のフォーマット。次のフォーマットがサポートされています： csv json avro debezium-json canal-json maxwell-json avro-confluent raw
   csv.field-delimiter	はい	CSV フォーマットのフィールド区切り文字。
   scan.startup.mode	はい	Kafka コンシューマーがデータの読み取りを開始するオフセットを指定します。有効な値： earliest-offset：利用可能な最小オフセットから読み取りを開始します。 latest-offset：最新のオフセットから読み取りを開始します。

4. 次の文を実行して、インポートジョブを作成します。

   INSERT INTO vector_ingest SELECT * FROM vector_kafka;