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. インデックスを作成します。

   index_params = milvus_client.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 タブで、root アカウントに対応するパスワードを確認します。
   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 を使用してターゲットコレクションの行数を確認します。