Milvus迁移至CSS向量数据库

应用场景

当您的业务初期使用Milvus进行纯向量相似度搜索，随着需求扩展需要同时具备全文检索和向量检索能力时，直接在Milvus上扩展全文检索功能成本高且架构复杂。通过将Milvus数据迁移至CSS向量数据库，可一站式满足混合检索需求，同时将向量搜索能力纳入已有的Elasticsearch技术栈，降低运维复杂度。

典型应用场景：

• 从向量检索向混合检索演进：业务需要同时具备全文检索（BM25/关键词匹配）和向量检索能力，实现一站式混合检索。
• 统一技术栈：企业已大规模使用Elasticsearch作为核心基础设施，希望将向量搜索能力也纳入Elasticsearch体系，降低运维复杂度。
• 向量检索与结构化过滤：业务场景中需要向量检索与大量标量字段的组合过滤（如时间范围、标签、分类等），CSS服务在倒排索引与向量索引的联合查询方面具有成熟支持。
• 平台统一管理：从Milvus平台迁移至华为云，实现平台统一管理。

方案架构

图1 Milvus迁移流程图

Milvus到CSS的迁移流程如下：

1. Schema分析：分析Milvus待迁移Collection的Schema（定义集合的字段结构和属性），设计CSS向量索引Mapping。
2. 数据导出：通过VTS（Vector Transport Service）或PyMilvus SDK从Milvus批量导出Collection数据。
3. 字段映射与转换：将Milvus字段类型映射为CSS向量数据库的字段类型，重点处理向量字段到vector类型的映射。
4. 索引创建：在CSS集群中创建向量索引，配置向量字段类型、维度、相似度算法等。
5. 数据写入：通过VTS直接写入数据，或使用Elasticsearch Bulk API批量写入数据。
6. 验证与切换：验证数据完整性，进行检索功能测试，确认无误后切换业务流量。

方案优势

• 一站式混合检索：迁移至CSS向量数据库后，可在同一引擎中实现全文检索与向量检索的混合查询，无需维护两套系统。
• 成熟的过滤能力：CSS向量数据库支持前置过滤、布尔查询等复合查询模式，向量检索与结构化过滤的联合查询性能优异。
• 灵活的索引算法：CSS向量数据库支持FLAT、GRAPH、GRAPH_PQ、IVF_GRAPH等多种向量索引算法，可根据数据规模和精度需求灵活选择。
• 批量写入优化：CSS向量数据库支持lazy_indexing延迟构建模式，海量数据导入时先写入后构建索引，显著提升写入吞吐量。

性能影响

• 源端Milvus影响：VTS或PyMilvus SDK批量读取Collection数据会对源端Milvus产生读取压力。建议在业务低峰期执行迁移，或通过调整“batch_size”参数控制读取速率。
• 目标端CSS集群影响：Elasticsearch Bulk API写入会消耗集群性能，写入期间建议将“refresh_interval”设为“-1”禁止自动刷新，或开启“lazy_indexing”延迟构建索引，写入完成后再构建索引。写入完成后恢复“refresh_interval”并触发索引构建。

约束限制

• 数据类型差异：Milvus的Array、JSON、VarChar等类型需映射为CSS的对应类型，部分复杂类型需要字段拆分或转换。Milvus与CSS的字段类型映射关系请参见表1。
• 向量字段映射：需确保Milvus与CSS的距离度量映射关系正确。Milvus与CSS的距离度量映射关系请参见表2。
• 网络互通：迁移执行机需与Milvus源端、CSS目标端网络互通。如涉及公网访问，需配置安全组或防火墙规则。
• 版本限制：仅Elasticsearch 7.10.2版本的CSS向量数据库支持作为Milvus迁移的目标端。
• 增量同步：迁移过程中Milvus的新增数据无法自动同步，建议在业务低峰期停写迁移。

资源和成本规划

步骤一：测试网络连通性

确认ECS与Milvus源端、CSS目标端网络互通。

1. 在ECS上执行以下命令，验证与Milvus的连通性。

   curl -ik http://<milvus_host>:<milvus_port>

   返回Milvus响应信息，表示网络连通。

2. 执行以下命令，验证与CSS集群的连通性。

   curl -u <es_username>:<es_password> -ik https://<css_host>:9200

   返回集群响应信息，表示网络连通。

步骤二：分析Milvus源端数据结构

查看Milvus集群待迁移Collection的结构信息，为后续设计CSS索引Mapping提供依据。

1. 在ECS上安装Milvus Python SDK依赖。

   pip install pymilvus

2. 连接Milvus集群，查看Collection列表，获取待迁移的Collection名称。

   from pymilvus import connections, utility
   
   connections.connect(host="<milvus_host>", port="<milvus_port>", token="<milvus_token>")
   collections = utility.list_collections()
   print(f"Collections: {collections}")

   表4 连接参数说明

   参数	说明
   milvus_host	Milvus服务内网IP地址。
   milvus_port	Milvus服务端口，如19530。
   milvus_token	Milvus认证令牌，未启用认证时省略该参数。配置示例：“root:Milvus”。

3. 查看目标Collection的Schema信息。

   from pymilvus import Collection
   
   collection = Collection("<collection_name>")
   schema = collection.schema
   
   # 查看主键字段
   print(f"Primary field: {schema.primary_field.name}, Type: {schema.primary_field.dtype}")
   
   # 查看所有字段
   for field in schema.fields:
       if field.dtype.name in ("FLOAT_VECTOR", "FLOAT16_VECTOR", "BFLOAT16_VECTOR"):
           dim = field.params.get("dim", "unknown")
           metric = field.params.get("metric_type", "unknown")
           print(f"  Vector Field: {field.name}, Type: {field.dtype.name}, Dim: {dim}, Metric: {metric}")
       else:
           print(f"  Scalar Field: {field.name}, Type: {field.dtype.name}")
   
   # 查看数据总量
   print(f"Entity count: {collection.num_entities}")

   表5 查询参数说明

   参数	说明
   collection_name	待迁移的Collection名称，如“quick_setup”。

4. 根据代码输出结果，记录以下关键信息，用于后续创建CSS向量索引。

   表6 关键信息与返回结果对应关系

   关键信息	返回结果来源	说明
   主键字段	Primary field行的“name”	Milvus的主键字段名，对应CSS索引的“ _id”或自定义主键字段。
   向量字段名称	Vector Field行的“name”	向量字段的名称，对应CSS索引中“type: vector”的字段。
   向量维度	Vector Field行的“Dim”	向量维度值，对应CSS索引中“dimension”参数。
   距离度量	Vector Field行的“Metric”	Milvus的距离度量（L2/IP/COSINE），需映射为CSS的metric参数，映射关系见表2。
   标量字段名称和类型	Scalar Field行的“name”和“Type”	标量字段，按表1映射为CSS字段类型。
   数据总量	Entity count	用于验证迁移完整性。

步骤三：创建CSS向量索引

根据Milvus Schema设计CSS向量索引Mapping，重点关注向量字段的映射。

1. 登录云搜索服务管理控制台。
2. 在左侧导航栏，选择“集群管理 > Elasticsearch”。
3. 在集群列表，选择本地集群，单击操作列的“Kibana”，登录Kibana。
4. 在Kibana左侧导航栏选择“Dev Tools”，进入操作页面。
5. 执行以下命令，基于步骤二：分析Milvus源端数据结构获取的关键信息创建向量索引。

   PUT /milvus_migrated_collection
   {
     "settings": {
       "index": {
         "vector": true,
         "number_of_shards": 3,
         "number_of_replicas": 1
       },
       "refresh_interval": "-1"
     },
     "mappings": {
       "properties": {
         "id": {
           "type": "long"
         },
         "title": {
           "type": "text",
           "fields": {
             "keyword": {
               "type": "keyword",
               "ignore_above": 256
             }
           }
         },
         "content": {
           "type": "text",
         },
         "category": {
           "type": "keyword"
         },
         "embedding": {
           "type": "vector",
           "dimension": 768,
           "indexing": true,
           "algorithm": "GRAPH",
           "metric": "euclidean"
         }
       }
     }
   }

   表7 向量索引关键参数说明列举了创建向量索引的关键参数说明，更全面的操作指导请参见创建向量索引。

   表7 向量索引关键参数说明

   参数	说明
   index.vector	必须设置为“true”，开启向量索引功能。
   refresh_interval	大批量写入时建议设为“-1”禁止自动刷新，写入完成后恢复。
   type	向量字段必须设置为“vector”。
   dimension	向量维度，必须与Milvus源端一致。取值范围1～4096。
   indexing	是否开启向量索引加速，设置为“true”后可使用VectorQuery查询。
   algorithm	向量索引算法，可选FLAT、GRAPH、GRAPH_PQ等。
   metric	距离度量方式，需与Milvus源端度量对应。

步骤四：执行数据迁移

根据实际环境选择以下方式之一执行数据迁移。

方式B：PyMilvus + Elasticsearch API脚本迁移

1. 在ECS上执行以下命令，安装迁移脚本的SDK依赖。

   pip install pymilvus elasticsearch numpy

2. 创建迁移配置文件“config.ini”。

   在ECS上执行vi config.ini，输入以下内容，执行wq保存为迁移配置文件。

   [Milvus]
   # Milvus服务主机地址
   host = <milvus_host>
   # Milvus服务端口
   port = <milvus_port>
   # 用户名
   username = '<milvus_username>'
   # 密码
   password = '<milvus_password>'
   # 数据库名称
   database = default
   # 待迁移的Collection名称
   collection = <collection_name>
   # 一批读取的文档数量，建议根据数据量调整（100～1000）
   batch_size = 1000
   # 主键字段名称，用于Elasticsearch文档ID
   primary_key = id
   # 向量字段
   vector_field = <vector_field_name>
   # 向量维度，为空则自动从Schema获取
   vector_dim = 
   # 文本字段名称，多个字段用逗号分隔，映射为text + keyword
   text_fields = 
   # 迁移的字段集合
   schema_fields = id,vector,color
   # Milvus迁移数据查询表达式，为空则迁移全部
   expr = 
   
   [Elasticsearch]
   # Elasticsearch连接地址，格式：host:port
   endpoint = https://<css_host>:9200
   # Elasticsearch用户名，仅安全集群涉及
   username = <es_username>
   # Elasticsearch密码，仅安全集群涉及
   password = <es_password>
   # 目标索引名称，为空则使用collection名称
   index_name = 
   # 目标索引分片数量
   number_of_shards = 3
   # 是否导出到文件
   export_file_flag = false

3. 创建迁移脚本“milvus2es.py”。

   在ECS上执行vi milvus2es.py，参考附录：迁移脚本编辑脚本内容，执行wq保存为迁移脚本。

4. 执行以下命令，启动迁移脚本，将Mivlus数据迁移到CSS。
   • 同时执行索引创建、数据导入和结果验证：

     python milvus2es.py

   • 通过“--step”参数指定执行步骤：

     # 创建索引：根据Milvus Collection的Schema在CSS中创建索引，包括字段类型映射和向量字段配置
     python milvus2es.py --step mapping
     
     # 导入数据：从Milvus分批读取数据并写入CSS，支持断点重试
     python milvus2es.py --step data
     
     # 验证结果:校验Milvus与CSS的文档数量是否一致，并抽样验证数据完整性
     python milvus2es.py --step verify

5. 启动迁移脚本的输出样例如下：

   2025-12-05 10:30:28,813 - INFO - Connected to Milvus Success:  2.6.5
   2025-12-05 10:30:29,092 - INFO - Loaded collection: quick_setup
   2025-12-05 10:30:29,092 - INFO - Collection fields : ['id', 'vector', 'color']
   2025-12-05 10:30:29,093 - INFO - Connected to Elasticsearch <bound method Elasticsearch.info of <Elasticsearch([{'host': '100.93.2.96', 'port': 9200, 'use_ssl': True}])>>
   2025-12-05 10:30:29,432 - INFO - HEAD https://100.93.2.96:9200/ [status:200 request:0.338s]
   2025-12-05 10:30:29,432 - INFO - 成功连接到Elasticsearch: https://100.93.2.96:9200
   2025-12-05 10:30:29,432 - INFO - Starting migration
   2025-12-05 10:30:29,504 - INFO - Milvus 集合实体总数: 490
   2025-12-05 10:30:29,575 - INFO - HEAD https://100.93.2.96:9200/quick_setup [status:200 request:0.070s]
   2025-12-05 10:30:29,575 - WARNING - ElasticSearch Index quick_setup exists
   2025-12-05 10:30:29,971 - INFO - POST https://100.93.2.96:9200/_bulk [status:200 request:0.079s]
   2025-12-05 10:30:31,211 - INFO - GET https://100.93.2.96:9200/quick_setup/_count [status:200 request:0.078s]
   2025-12-05 10:30:31,211 - INFO - Milvus文档数: 490, Elasticsearch文档数: 10
   2025-12-05 10:30:31,211 - WARNING - 文档数量不匹配: Milvus=490, Elasticsearch=10
   2025-12-05 10:30:31,211 - INFO - Migration completed in 0:00:01.778907

步骤五：验证数据完整性

迁移完成后，确认CSS集群中的数据与Milvus源端一致。

登录Elasticsearch集群的Kibana，在"Dev Tools"中执行以下命令，查看迁移数据量：

1. 登录Elasticsearch集群的Kibana。
2. 在Kibana左侧导航栏选择“Dev Tools”，进入操作页面。
3. 执行以下命令，查看迁移数据量。

   GET /milvus_migrated_collection/_count

   milvus_migrated_collection要替换为目标索引名称。

4. 对比Milvus源端的数据总量（步骤二：分析Milvus源端数据结构获取的Entity count）与CSS目标端的文档数量是否一致。
5. （可选）如果创建向量索引时使用了lazy_indexing，需要先执行离线构建再执行VectorQuery标准查询。

   POST _vector/indexing/milvus_migrated_collection
   {
     "field": "embedding"
   }

6. 执行向量检索，验证检索功能是否正常。

   POST /milvus_migrated_collection/_search
   {
     "size": 5,
     "query": {
       "vector": {
         "embedding": {
           "vector": [0.1, 0.2, ...],
           "topk": 5
         }
       }
     }
   }

7. （可选）如果迁移过程中禁止自动刷新索引，则迁移验证通过后，需要恢复索引的refresh_interval设置。

   PUT /milvus_migrated_collection/_settings
   {
     "refresh_interval": "1s"
   }

常见问题

• 问题1：迁移后CSS向量检索结果与Milvus不一致怎么办？

  可能原因和解决方法：

  • 距离度量不匹配：确认Milvus和CSS使用了相同的距离度量，参考表2的映射关系。
  • HNSW参数差异：CSS和Milvus的HNSW参数（neighbors、efc、ef等）不同，会影响召回结果。可通过调整CSS的向量索引参数优化。
  • 浮点精度：向量数据在导出或导入过程中可能存在浮点精度损失，建议使用double类型存储中间结果。
• 问题2：亿级数据量的迁移时间过长怎么办？
  • 分批迁移：按Milvus分区（Partition）或主键范围分批迁移，降低单次任务压力。
  • 并行迁移：启动多个VTS实例或脚本进程，分别迁移不同的Collection或Partition。
  • 增大CSS集群规模：临时扩容CSS集群，提升写入和索引构建速度。
  • 开启延迟构建：在CSS向量索引中设置lazy_indexing为"true"，全量数据写入后再触发离线构建。
• 迁移过程中Milvus有新数据写入怎么办？

  处理建议：

  • 停写迁移：在业务低峰期暂停Milvus写入，完成迁移后切换至CSS。
  • 双写方案：在迁移期间，应用层同时写入Milvus和CSS，迁移完成后停止Milvus写入。
  • 增量补偿：记录迁移开始时间戳，迁移完成后通过Milvus的query接口查询该时间戳之后的新增数据，补偿写入CSS。

附录：迁移脚本

使用PyMilvus + Elasticsearch API脚本方式迁移时，需要创建迁移脚本“milvus2es.py”，内容示例：

import argparse
import configparser
import json
import logging
import time
from datetime import datetime
from typing import Dict, Any, List
from typing import Optional
import numpy as np
import urllib3
from elasticsearch import Elasticsearch
from elasticsearch import helpers
from pymilvus import connections, Collection, utility
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('migration.log'), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)
def load_config(config_file: str) -> configparser.ConfigParser:
    config = configparser.ConfigParser()
    try:
        if not config.read(config_file, encoding='utf-8'):
            raise FileNotFoundError(f"Config file {config_file} not found")
        return config
    except Exception as e:
        logger.error(f"Failed to load config: {e}")
        raise


# 导入数据到Elasticsearch
def es_run_once(es_client, actions, count):
    try:
        for success, info in helpers.parallel_bulk(
                es_client, actions, chunk_size=100, queue_size=4, request_timeout=60):
            if not success:
                print('A document failed: ', info)
        print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}: {count}')
        time.sleep(1)
    except Exception as err:
        print(f"bulk Elasticsearch error {err}")
        try:
            for success, info in helpers.parallel_bulk(
                    es_client, actions, chunk_size=20, queue_size=2, request_timeout=60):
                if not success:
                    print('A document failed: ', info)
            print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}: retry {count}')
        except Exception as err:
            print(f"retry error {err}")
class MilvusToESMigrator:
    def __init__(self, config_file: str = 'config.ini'):
        self.config = load_config(config_file)
        # milvus
        self.milvus_host = self.config.get('Milvus', 'host')
        self.milvus_port = self.config.get('Milvus', 'port')
        self.milvus_username = self.config.get('Milvus', 'username')
        self.milvus_password = self.config.get('Milvus', 'password')
        self.database = self.config.get('Milvus', 'database')
        self.collection_name = self.config.get('Milvus', 'collection')
        self.batch_size = self.config.getint('Milvus', 'batch_size', fallback=100)
        self.primary_key = self.config.get('Milvus', 'primary_key')
        self.vector_field = self.config.get('Milvus', 'vector_field')
        str_vector_dim = self.config.get('Milvus', 'vector_dim', fallback=None)
        if str_vector_dim:
            self.vector_dim = int(str_vector_dim)
        else:
            self.vector_dim = None
        self.text_fields = self._parse_text_fields(self.config.get('Milvus', 'text_fields'))
        self.schema_fields = self._parse_text_fields(self.config.get('Milvus', 'schema_fields'))
        self.expr = self.config.get('Milvus', 'expr', fallback=None)
        # elasticsearch
        self.es_host = self.config.get('ElasticSearch', 'endpoint')
        self.es_username = self.config.get('ElasticSearch', 'username')
        self.es_password = self.config.get('ElasticSearch', 'password')
        index_name = self.config.get('ElasticSearch', 'index_name')
        self.es_index_name = self.collection_name if not index_name else index_name
        self.number_of_shards = self.config.getint('ElasticSearch', 'number_of_shards')
        self.export_file_flag = self.config.getboolean('ElasticSearch', 'export_file_flag', fallback=False)
        self.fields = []
        self.MAX_SPARSE_DIMENSION = 1000
        self.milvus_version = None

        # Connect to Milvus
        self.collection = self._connect_milvus()
        # Connect to es client
        self.es_client = self._connect_es()
    def _parse_text_fields(self, text_fields_str: str) -> List[str]:
        if not text_fields_str:
            return []
        return [field.strip() for field in text_fields_str.split(',')]
    def _connect_milvus(self) -> Collection:
        try:
            connections.connect(
                db_name=self.database,
                user=self.milvus_username,
                password=self.milvus_password,
                host=self.milvus_host,
                port=self.milvus_port
            )
            self.milvus_version = utility.get_server_version()
            logger.info(f"Connected to Milvus Success:  {self.milvus_version}")
            collection = Collection(self.collection_name)
            collection.load()
            logger.info(f"Loaded collection: {self.collection_name}")
            fields = [field.name for field in collection.schema.fields]
            logger.info(f"Collection fields : {fields}")
            return collection
        except Exception as e:
            logger.error(f"Milvus connection failed: {e}")
            raise

    def _connect_es(self) -> Elasticsearch:
        try:
            auth = (self.es_username, self.es_password)
            es = Elasticsearch(self.es_host, http_auth=auth, verify_certs=False, max_retries=10,
                               retry_on_timeout=True)
            logger.info(f"Connected to Elasticsearch {es.info}")
            # 测试连接
            if es.ping():
                logger.info(f"成功连接到Elasticsearch: {self.es_host}")
            else:
                raise Exception("Elasticsearch连接测试失败")
            return es
        except Exception as e:
            logger.error(f"Elasticsearch connection failed: {e}")
            raise

    @staticmethod
    def _milvus_to_es_type(milvus_type: str, dim: Optional[int] = None) -> str:
        type_map = {
            "INT64": "long",
            "INT32": "integer",
            "INT16": "integer",
            "INT8": "integer",
            "FLOAT": "float",
            "DOUBLE": "double",
            "BOOL": "boolean",
            "VARCHAR": "keyword",
            "STRING": "text",
            "JSON": "object"
            # 向量类型在Elasticsearch中有专门处理，这里只处理基础类型
        }
        return type_map.get(milvus_type, "text")
    def __create_es_index(self, fields_mapping: Dict[str, Any]) -> bool:
        try:
            # 检查索引是否存在
            if self.es_client.indices.exists(index=self.es_index_name):
                logger.warning(f"索引 {self.es_index_name} 已存在")
                return True

            # 构建索引映射
            mapping = {
                "settings": {
                    "index": {
                        "vector": "true",  # 开启向量特性
                        "number_of_shards": self.number_of_shards,  # 分片数据量
                        "number_of_replicas": 1,
                        "refresh_interval": "1s"
                    }
                },
                "mappings": {  # 添加Mapping层
                    "properties": {}  # 添加properties层
                }
            }
            for field_name, field_config in fields_mapping.items():
                mapping["mappings"]["properties"][field_name] = field_config
            # 创建索引
            self.es_client.indices.create(index=self.es_index_name, body=mapping)
            logger.info(f"成功创建Elasticsearch索引: {self.es_index_name},  mapping {mapping}")
            return True
        except Exception as e:
            logger.error(f"创建Elasticsearch索引失败: {e}")
            return False

    def create_elasticsearch_index(self) -> None:
        try:
            total_entities = self.collection.num_entities
            logger.info(f"Milvus 集合实体总数: {total_entities}")
            # Check if table exists
            if self.es_client.indices.exists(index=self.es_index_name):
                logger.warning(f"ElasticSearch Index {self.es_index_name} exists")
                return

            # 获取集合schema
            schema = self.collection.schema
            # 构建Elasticsearch字段映射
            fields_mapping = {}
            for field in schema.fields:
                field_name = field.name
                if field_name in self.text_fields:
                    fields_mapping[field_name] = {
                        "type": "text",
                        "fields": {
                            "keyword": {"type": "keyword", "ignore_above": 256}
                        }
                    }
                elif field_name == self.vector_field:
                    vector_dim = self.vector_dim
                    if not vector_dim:
                        vector_dim = field.dim if hasattr(field, 'dim') else None
                        vector_dim = field.params['dim'] if vector_dim is None and 'dim' in field.params else vector_dim
                    fields_mapping[field_name] = {
                        "type": "vector",
                        "dimension": vector_dim,
                        "indexing": True,
                        "algorithm": "GRAPH",
                        "metric": "cosine"
                    }
                else:
                    # 其他字段根据类型映射
                    fields_mapping[field_name] = {"type": self._milvus_to_es_type(field.dtype.name)}
            # Build CREATE INDEX
            self.__create_es_index(fields_mapping)
            logger.info(f"Created index: {self.es_index_name}")
        except Exception as e:
            logger.error(f"Table creation failed: {e}")
            raise

    def __export_partition_data(self, partition_name):
        query_iter = self.collection.query_iterator(
            batch_size=self.batch_size,
            partition_names=[partition_name],
            output_fields=["id"] if not self.schema_fields else self.schema_fields,
            expr=self.expr
        )
        export_path = None
        if self.export_file_flag:
            if partition_name == '_default':
                export_path = f'{self.es_index_name}_default.txt'
            else:
                export_path = f'{self.es_index_name}_{partition_name}.txt'

        while True:
            results = query_iter.next()
            if len(results) == 0:
                # close the iterator
                query_iter.close()
                break

            # 准备Elasticsearch批量操作
            actions = []
            for entity in results:
                doc_id = entity.get(self.primary_key)
                # 构建文档
                doc_source = {}
                for field in self.text_fields:
                    if field in entity:
                        doc_source[field] = entity[field]
                # 处理向量字段
                if self.vector_field in entity:
                    vector_data = entity[self.vector_field]
                    if isinstance(vector_data, np.ndarray):
                        doc_source[self.vector_field] = vector_data.tolist()
                    else:
                        doc_source[self.vector_field] = vector_data
                # 添加其他字段
                for key, value in entity.items():
                    if key not in self.text_fields + [self.vector_field]:
                        doc_source[key] = value
                action = {
                    "_index": self.es_index_name,
                    "_id": str(doc_id),
                    "_source": doc_source
                }
                actions.append(action)
            if export_path:
                with open(export_path, 'w', encoding='utf-8') as export_file:
                    json.dump(actions, export_file)
                    export_file.write('\n')
            es_run_once(self.es_client, actions, 100)
    def import_data(self) -> None:
        for partition in self.collection.partitions:
            partition_name = partition.name
            self.__export_partition_data(partition_name)
    def verify_migration(self, sample_size: int = 100) -> bool:
        try:
            # 检查文档数量， 如果milvus中重复primary_key更新, 会导致统计值不一致
            milvus_count = self.collection.num_entities
            es_count = self.es_client.count(index=self.es_index_name)['count']
            logger.info(f"Milvus文档数: {milvus_count}, Elasticsearch文档数: {es_count}")
            if milvus_count != es_count:
                logger.warning(f"文档数量不匹配: Milvus={milvus_count}, Elasticsearch={es_count}")
                return False
            # 抽样验证数据一致性
            results = self.collection.query(expr="", limit=sample_size)
            for entity in results:
                doc_id = entity.get('id')
                if doc_id:
                    es_doc = self.es_client.get(index=self.es_index_name, id=str(doc_id), ignore=[404])
                    if not es_doc['found']:
                        logger.warning(f"文档 {doc_id} 在Elasticsearch中不存在")
                        return False

            logger.info("迁移验证通过!")
            return True
        except Exception as e:
            logger.error(f"验证迁移结果失败: {e}")
            return False

    def run_migration(self, steps: Optional[List[str]] = None) -> None:
        start_time = datetime.now()
        if steps is None:
            steps = ['mapping', 'data', 'verify']
        logger.info(f"Starting migration, steps: {steps}")
        try:
            if 'mapping' in steps:
                self.create_elasticsearch_index()
            if 'data' in steps:
                self.import_data()
            if 'verify' in steps:
                self.verify_migration()
            logger.info(f"Migration completed in {datetime.now() - start_time}")
        except Exception as e:
            logger.error(f"Migration failed: {e}")
            raise


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Milvus to Elasticsearch migration tool")
    parser.add_argument('--step', nargs='*', choices=['mapping', 'data', 'verify'],
                        default=['mapping', 'data', 'verify'],
                        help='指定执行的步骤，可选: mapping, data, verify。不指定则全部执行')
    parser.add_argument('--config', type=str, default='config.ini',
                        help='配置文件路径，默认: config.ini')
    args = parser.parse_args()
    try:
        migrator = MilvusToESMigrator(config_file=args.config)
        migrator.run_migration(steps=args.step if args.step else None)
    except Exception as e:
        logger.error(f"Migration failed: {e}")
        exit(1)

相关文档

• CSS向量数据库：了解CSS向量数据库的使用方法，包括创建向量索引、导入向量数据、进行向量检索等。
• 使用Elasticsearch实现向量检索：了解如何快速上手CSS向量检索功能。
• VTS：了解VTS迁移工具的使用方法。
• PyMilvus SDK：了解Milvus Python SDK的安装和使用方法。