Vector Retrieval Service for Milvus:イメージツールを使用した Alibaba Cloud Milvus へのデータ移行

ネットワーク要件

コンテナーをデプロイするネットワークから、ソースの Milvus インスタンスと宛先の Alibaba Cloud Milvus インスタンスの両方にアクセスできる必要があります。最適なパフォーマンスを得るには、宛先インスタンスと同じ VPC にコンテナーをデプロイしてください。

1. バックグラウンドでコンテナーを起動します。

   docker run -d -it \
     --name milvus-migration \
     registry.cn-hangzhou.aliyuncs.com/taihao-executor/taihao-executor:release_2.22.0-ali \
     /bin/bash

2. コンテナー ID を表示し、コンテナーにアクセスします。

   # コンテナーを照会
   docker ps
   
   # コンテナーに入る (実際のコンテナー ID に置き換えてください)
   docker exec -it <container_id> bash

   例：

   docker exec -it 55ac98f3b054 bash

設定テンプレート

hoconenv {
  parallelism = 1           # 同時実行数。初期値を 1 に設定します。
  job.mode = "BATCH"        # バッチモード。
}

source {
  Milvus {
    url = "http://<source_instance_address>:19530"       # 内部ネットワークアドレスをサポートしています。
    token = "<username>:<password>"                 # 例：root:Test123456@
    database = "default"                    # デフォルトは "default" です。list_databases を実行して他のデータベースを照会できます。
    collections = ["col_a", "col_b"]        # 移行するコレクションのリスト。
    batch_size = 10000                      # 一度に読み取るエントリ数。大きなテーブルの場合はこの値を増やすことができます。
  }
}

sink {
  Milvus {
    url = "http://<destination_alibaba_cloud_milvus_address>:19530"
    token = "<destination_instance_token>"
    database = "default"
    batch_size = 1000
    enable_auto_id = false                 # ソースコレクションに自動増分 ID がある場合は false に設定します。それ以外の場合は true に設定します。
  }
}

注意事項

• ソースコレクションのロード：移行したいすべてのコレクションを load() メソッドを使用してロードする必要があります。そうしないと、エラーが発生します。

• すべてのコレクションを移行する場合：設定ファイルから collections の行を削除すると、ロードされているすべてのコレクションが自動的に同期されます。

• 内部ネットワークアドレスの使用：コンテナーと宛先インスタンスが同じリージョンにある場合、データ転送速度を向上させるために、宛先インスタンスの内部ネットワークエンドポイントを使用してください。

方法 1：ローカルモード (単一マシンでの操作)

nohup ./bin/seatunnel.sh --config ./migration.conf -m local > migration.log 2>&1 &

-Xms4g
-Xmx8g

方法 2：クラスターモード (高性能向けに推奨)

大量のデータを移行する場合に適しています：

# ログディレクトリを作成
mkdir -p ./logs

# クラスターサービスを開始
./bin/seatunnel-cluster.sh -d

# タスクを送信
nohup ./bin/seatunnel.sh --config ./migration.conf > migration.log 2>&1 &

1. インデックスを作成します。

   milvus_client = milvus.prepare_index_params()
   index_params.add_index(
           field_name="vector",  # インデックスを作成するベクターフィールドの名前
           index_type="HNSW",  # 作成するインデックスのタイプ
           index_name="vector_index",  # 作成するインデックスの名前
           metric_type="L2",  # 類似性を測定するために使用されるメトリックタイプ
           params={
               "M": 64,  # 各ノードがグラフ内で接続できる隣接ノードの最大数
               "efConstruction": 100  # インデックス構築中に接続候補として考慮される隣接ノードの数
           }  # インデックス構築パラメーター
       )
   milvus_client.create_index("collectionName", index_params)

2. コレクションをメモリにロードします。

   milvus_client.load_collection()

   コレクションをロードする前にインデックスを作成してください。そうしないと、高速検索が有効になりません。主要なパラメーター：

   パラメーター	取得方法
   url	Alibaba Cloud Milvus コンソールにログインします。Security Configuration タブで、パブリックネットワークアドレスまたは内部ネットワークアドレスを表示します。パフォーマンスを向上させるために、内部ネットワークアドレスを使用することをお勧めします。
   token	フォーマットは username:password です。例: root:YourPassword123@。Alibaba Cloud Milvus コンソールにログインします。Security Configuration タブで、ルートアカウントのパスワードを確認します。
   database	デフォルトは `default` です。複数データベース機能を使用している場合、list_databases() API を使用して他のデータベースを照会できます。

   完全な設定：

   env {
     parallelism = 1
     job.mode = "BATCH"
   }
   
   source {
     Milvus {
       url = "http://xx.xx.xx.xx:19530"
       token = "root:SourcePass123@"
       database = "default"
       collections = ["medium_articles"]
       batch_size = 10000
     }
   }
   
   sink {
     Milvus {
       url = "http://proxy-bj.vpc.milvus.aliyuncs.com:19530"
       token = "root:TargetPass123@"
       database = "default"
       batch_size = 10000
       enable_auto_id = false
     }
   }

Q1：移行中に「Collection not loaded」というエラーが発生します。どうすればよいですか？

A：移行対象のすべてのソースコレクションが .load() メソッドを使用してメモリにロードされていることを確認してください。

Q2：特定のフィールドのみを移行できますか？

A：いいえ、できません。現在のバージョンでは、コレクション全体の移行のみがサポートされています。フィールドのフィルタリングはサポートされていません。

Q3：移行の進捗状況を監視するにはどうすればよいですか？

A：migration.log ファイルの出力を確認できます。また、Attu を使用して宛先コレクションの行数の変化を監視することもできます。