PolarDB:PolarSearch へのデータベクトル化と同期の自動化

仕組み

自動化されたデータワークフローには、PolarDB for MySQL、PolarSearch、および埋め込みモデルサービスが含まれます。主要なワークフローは以下のとおりです。

1. データソース：元のテキストデータは PolarDB for MySQL テーブルに格納されます。

2. 同期のトリガー：PolarDB for MySQL の Search View を呼び出して、MySQL から PolarSearch へのデータ同期タスク (AutoETL) を作成し、開始します。

3. データの書き込み：AutoETL タスクは、テーブルのデータ変更 (挿入、削除、更新) を、指定された PolarSearch インデックスにリアルタイムで同期します。

4. パイプライン処理：データが PolarSearch インデックスに書き込まれる前に、PolarSearch は事前定義されたインジェストパイプラインをトリガーします。

5. モデルの呼び出し：パイプライン内の text_embedding プロセッサーは、指定されたテキストフィールドのコンテンツを読み取り、コネクターを介して外部の埋め込みモデルサービスを呼び出し、テキストをベクトルに変換します。

6. ベクトルの格納：パイプラインは、モデルから返されたベクトルを元のデータとともに PolarSearch インデックスの対応するフィールドに書き込み、ベクトルインデックスを完成させます。

前提条件

この機能を使用する前に、お使いの環境が以下の要件を満たしていることを確認してください。

• クラスターバージョン：

  • MySQL 8.0.1、リビジョン 8.0.1.1.52 以降。

  • MySQL 8.0.2、リビジョン 8.0.2.2.33 以降。

• ネットワーク環境：テキストのベクトル化に使用される埋め込みモデルサービスは、PolarDB for MySQL クラスターと同じ VPC 内にデプロイされている必要があります。PolarSearch ノードは、クライアントとしてこのモデルサービスに接続します。

重要な注意事項

更新シナリオでインジェストパイプラインを使用してテキストデータを自動的にベクトル化する場合、Search View はデフォルトで更新書き込みモード (Update) を使用して PolarSearch のデータを変更します。このモードは変更されたフィールドのみを更新し、対応するベクトルフィールドは自動的に更新しません。

ソーステーブルのテキストフィールドを更新した後に、対応するベクトルを自動的に再生成する必要がある場合は、Search View が インデックス書き込みモード (Index) を使用するように設定できます。Search View を作成する前に、次のコマンドを実行してください。

SET esl_sink_options = "'sink.force-index-request' = 'true'";

インデックス書き込みモードを有効にすると、各同期は変更されたフィールドのみを更新するのではなく、PolarSearch インデックス内のドキュメント全体を置き換えます。

埋め込みモデルサービスの準備

PolarDB for MySQL を設定する前に、HTTP でアクセス可能な埋め込みモデルサービスを準備し、PolarSearch からアクセスできることを確認してください。このサービスは、テキストを入力として受け取り、対応するベクトルを返します。デプロイの詳細な手順については、「外部モデルサービスの統合」をご参照ください。

アクセス認証情報および環境変数の設定

開始前に、以下の情報を準備し、環境変数として設定してください。これらの値を一元管理することで、以降の curl コマンドが簡潔になり、コピーおよび実行が容易になります。

手順：ターミナルで以下のコマンドを実行し、例の値を実際の情報に置き換えてください。

# PolarSearch のホストおよびポートを設定
export POLARSEARCH_HOST_PORT="pc-xxx.polardbsearch.rds.aliyuncs.com:3001"

# PolarSearch の管理者ユーザー名およびパスワードを設定
export USER_PASSWORD="polarsearch_user:your_password"

# Alibaba Cloud Model Studio (Bailian) の API キーを設定
export YOUR_API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx"

インジェストパイプラインの作成

データがインデックスに書き込まれる前に自動的にベクトル化を実行するインジェストパイプラインを作成します。text_embedding プロセッサーを使用する text_to_vec_pipeline という名前のパイプラインを定義します。

パラメーター

• model_id：「外部モデルサービスの統合」でデプロイした埋め込みモデルの ID です。

• field_map：入力フィールドと出力フィールド間のマッピングを定義します。たとえば、マッピングは my_text フィールドからテキストコンテンツを読み取り、モデルを呼び出してベクトルを生成し、そのベクトルの結果を my_vector フィールドに格納します。

curl -XPUT "http://${POLARSEARCH_HOST_PORT}/_ingest/pipeline/text_to_vec_pipeline" \
--user "${USER_PASSWORD}" \
-H 'Content-Type: application/json' \
-d '{
  "description": "A text embedding pipeline",
  "processors": [
    {
      "text_embedding": {
        "model_id": "<your-deployed-embedding-model-id>",
        "field_map": {
          "my_text": "my_vector"
        }
      }
    }
  ]
}'

PUT _ingest/pipeline/text_to_vec_pipeline
{
  "description": "A text embedding pipeline",
  "processors": [
    {
      "text_embedding": {
        "model_id": "<your-deployed-embedding-model-id>",
        "field_map": {
          "my_text": "my_vector"
        }
      }
    }
  ]
}

データ同期の設定

1. テストデータの準備：PolarDB for MySQL クラスターにログインし、データベースとテーブルを作成して、サンプルデータを挿入してください。

   CREATE DATABASE IF NOT EXISTS db;
   CREATE TABLE IF NOT EXISTS db.test_table (
       id INT PRIMARY KEY, 
       t1 INT, 
       t2 TEXT
   );
   INSERT INTO db.test_table(id, t1, t2) VALUES 
   (1, 11, 'aaa'), 
   (2, 22, 'bbb'), 
   (3, 33, 'ccc');

2. インデックスの作成：元のデータとベクトルを格納するための PolarSearch インデックスを作成してください。

   • default_pipeline：このインデックスのデフォルトパイプラインを、前のステップで作成した text_to_vec_pipeline に設定します。これにより、インデックスに書き込まれるすべてのデータが自動的にベクトル化されます。

   • my_vector.type：ベクトルフィールドのデータ型は knn_vector である必要があります。

   • my_vector.dimension：ベクトルのディメンションは、モデルの実際の出力ディメンション (1024) と完全に一致する必要があります。

   CLI

   curl -XPUT "http://${POLARSEARCH_HOST_PORT}/test_index" \
   --user "${USER_PASSWORD}" \
   -H 'Content-Type: application/json' \
   -d '{
     "settings": {
       "index": {
         "knn": true,
         "default_pipeline": "text_to_vec_pipeline"
       }
     },
     "mappings": {
       "properties": {
         "id": {
           "type": "integer"
         },
         "my_text": {
           "type": "text"
         },
         "my_vector": {
           "type": "knn_vector",
           "dimension": 1024,
           "method": {
             "engine": "faiss",
             "name": "hnsw"
           }
         }
       }
     }
   }'

   Dashboard

   PUT /test_index
   {
     "settings": {
       "index": {
         "knn": true,
         "default_pipeline": "text_to_vec_pipeline"
       }
     },
     "mappings": {
       "properties": {
         "id": {
           "type": "integer"
         },
         "my_text": {
           "type": "text"
         },
         "my_vector": {
           "type": "knn_vector",
           "dimension": 1024,
           "method": {
             "engine": "faiss",
             "name": "hnsw"
           }
         }
       }
     }
   }

3. AutoETL 同期タスクの作成と開始：PolarDB for MySQL クラスターで Search View を作成してください。

   このコマンドは、db.test_table から test_index へのマッピングを作成します。db.test_table のデータが変更されると、AutoETL は id フィールドと t2 フィールドの値を、それぞれ test_index の id フィールドと my_text フィールドに同期します。test_index にはデフォルトのパイプラインがあるため、データを書き込むと自動的にベクトル化プロセスがトリガーされ、結果のベクトルは my_vector フィールドに格納されます。

   CREATE SEARCH VIEW test_index(id, my_text) AS SELECT id, t2 FROM db.test_table;

データ同期の検証

PolarSearch の test_index をクエリして、データが書き込まれ、my_vector フィールドにベクトルデータが含まれていることを確認してください。

curl -XPOST "http://${POLARSEARCH_HOST_PORT}/test_index/_search" \
--user "${USER_PASSWORD}" \
-H 'Content-Type: application/json' \
-d '{
  "query": {
    "match": {
      "my_text": "aaa"
    }
  }
}'

POST /test_index/_search
{
  "query": {
    "match": {
      "my_text": "aaa"
    }
  }
}

期待されるレスポンスは以下のとおりです (抜粋)：

"hits": [
      {
        "_index": "test_index",
        "_id": "1",
        "_score": 0.44583148,
        "_source": {
          "my_text": "aaa",
          "id": 1,
          "my_vector": [
            -0.013453668,
            0.009771001,
            -0.00977745,
            ...
          ]
        }
      }
    ]