Vector Retrieval Service for Milvus:イメージツールを使用した Alibaba Cloud 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

設定テンプレート

env {
  parallelism = 1           # 同時実行レベル。初期値として 1 に設定することを推奨します。
  job.mode = "BATCH"        # バッチ処理モード。
}

source {
  Milvus {
    url = "http://<source-instance-endpoint>:19530"       # 内部エンドポイントがサポートされています。
    token = "<username>:<password>"                 # 例：root:Test123456@
    database = "default"                    # 移行元のデータベース。デフォルトは `default` です。
    collections = ["col_a", "col_b"]        # 移行するコレクションのリスト。
    batch_size = 10000                      # バッチごとに読み取るエンティティ数。大規模なコレクションの場合は、この値を増やしてください。
  }
}

sink {
  Milvus {
    url = "http://<target-Alibaba-Cloud-Milvus-endpoint>:19530"
    token = "<target-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：移行の進捗は 2 つの方法で監視できます。migration.log ファイルの出力を確認するか、Attu を使用してターゲットコレクションの行数を監視します。