Migrating Data from Milvus to the CSS Vector Database

Scenarios

If your application initially uses Milvus for pure vector similarity search but later requires both full-text search and vector search, extending Milvus to support full-text search can be costly and architecturally complex. By migrating data from Milvus to the CSS vector database, you can implement hybrid search in a single platform while integrating vector search into your existing Elasticsearch technology stack, reducing operational complexity.

Typical scenarios include:

• Evolution from vector search to hybrid search: Services need both full-text search (BM25/keyword matching) and vector search capabilities to implement one-stop hybrid search.
• Unified technology stack: Organizations that already use Elasticsearch extensively as their core search infrastructure want to integrate vector search into the existing technology stack to simplify operations and maintenance.
• Combining vector search with structured filtering: Applications require vector search along with extensive filtering on scalar fields (such as time ranges, tags, or categories). CSS provides mature support for combining inverted indexes with vector indexes in joint queries.
• Unified platform management: Migrate from a self-managed or third-party Milvus deployment to Huawei Cloud CSS for centralized platform management.

Solution Architecture

Figure 1 Milvus migration procedure

The procedure for migrating from Milvus to CSS is as follows:

1. Analyze the Milvus collection schema. Analyze the schema of the Milvus collection to be migrated and design a corresponding CSS vector index mapping.
2. Export data. Export collection data from Milvus in bulk using VTS or the PyMilvus SDK.
3. Map and convert fields. Map Milvus field types to their corresponding CSS vector database field types, while paying particular attention to mapping vector fields to the CSS vector type.
4. Create a vector index. Create a vector index in the CSS cluster and configure the vector field type, dimension, similarity metric, and indexing algorithm.
5. Import data. Import data either directly through VTS or in bulk using the Elasticsearch Bulk API.
6. Validate and switch over. Verify data integrity, test search functionality, and switch production traffic after validation is complete.

Advantages

• One-stop hybrid search: After migration to the CSS vector database, both full-text search and vector search can be performed within a single engine, so there is no need to maintain separate systems.
• Advanced filtering capabilities: The CSS vector database supports compound queries such as pre-filtering and Boolean queries, delivering excellent performance by combining vector search with structured filtering.
• Flexible indexing algorithms: The CSS vector database supports multiple vector indexing algorithms, including FLAT, GRAPH, GRAPH_PQ, and IVF_GRAPH, allowing you to balance search performance and recall based on your workload.
• Optimized bulk ingestion: The CSS vector database supports the lazy_indexing mode, which defers vector index creation until after data import, significantly improving bulk ingestion throughput.

Performance Impact

• Impact on the source Milvus cluster: Reading collection data in bulk through VTS or the PyMilvus SDK increases the read load on the source Milvus cluster. Perform this operation during off-peak hours, or reduce the read rate by adjusting the batch_size parameter.
• Impact on the destination CSS cluster: Bulk writes through the Elasticsearch Bulk API consume cluster resources. To mitigate this impact during data import, set refresh_interval to -1 to disable automatic refreshes, or enable lazy_indexing to defer vector index creation until after the data import is complete. Once the import is complete, restore the refresh_interval setting and trigger index creation.

Constraints

• Data type differences: Milvus data types such as Array, JSON, and VarChar must be mapped to their corresponding CSS data types. Some complex types may require field splitting or conversion. For details, see Table 1.
• Vector field mapping: Ensure that the distance metrics used by Milvus are correctly mapped to their CSS equivalents. For details, see Table 2.
• Network connectivity: The PC where you are performing the migration must be able to communicate with both the source Milvus cluster and the destination CSS cluster. If public network access is used, configure the required security group or firewall rules.
• Version requirements: Only CSS vector databases running Elasticsearch 7.10.2 can be used as the destination for migrating data from Milvus.
• Incremental synchronization: New data written to Milvus after the migration has already started is not synchronized automatically. You are advised to stop writes during migration and perform the migration during off-peak hours.

Resource and Cost Planning

Step 1: Test Network Connectivity

Verify that the ECS can communicate with both the source Milvus cluster and the destination CSS cluster.

1. Run the following command on the ECS to verify connectivity to the Milvus cluster.

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

   If Milvus returns a response, the network is connected.

2. Run the following command to verify connectivity to the CSS cluster.

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

   If the cluster returns a response, the network is connected.

Step 2: Analyze the Source Data Structure in Milvus

Examine the schema of the Milvus collection to be migrated. The collected information will be used to design the CSS vector index mapping.

1. Install the Milvus Python SDK on the ECS.

   pip install pymilvus

2. Connect to the Milvus cluster and list all collections to identify the collection to migrate.

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

   Table 4 Connection parameters

   Parameter	Description
   milvus_host	Private IP address of the Milvus service.
   milvus_port	Milvus service port, for example, 19530.
   milvus_token	Authentication token for Milvus. Omit this parameter if authentication is disabled. Example: root:Milvus.

3. View the schema of the target collection.

   from pymilvus import Collection
   
   collection = Collection("<collection_name>")
   schema = collection.schema
   
   # View the primary key field.
   print(f"Primary field: {schema.primary_field.name}, Type: {schema.primary_field.dtype}")
   
   # View all fields.
   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}")
   
   # View the total number of entities.
   print(f"Entity count: {collection.num_entities}")

   Table 5 Query parameter

   Parameter	Description
   collection_name	Name of the collection to migrate, for example, quick_setup.

4. Record the following information from the command output. This information will be used when creating the CSS vector index.

   Table 6 Key information and fields in the command output

   Information	Command Output Field	Description
   Primary key field	name in the Primary field output	The primary key field in Milvus. It maps to the _id field or a custom primary key field in the CSS index.
   Vector field name	name in the Vector Field output	Name of the vector field. It maps to a field of type: vector in the CSS vector index.
   Vector dimension	Dim in the Vector Field output	Vector dimension. It corresponds to the dimension parameter in the CSS vector index.
   Distance metric	Metric in the Vector Field output	Distance metric used by Milvus (L2/IP/COSINE). It must be mapped to the corresponding metric in CSS. For details, see Table 2.
   Scalar field names and types	name and Type in the Scalar Field output	Scalar fields. Map them to the corresponding CSS field types according to Table 1.
   Total entity count	Entity count	Used to verify that all data has been migrated.

Step 3: Create a CSS Vector Index

Create a CSS vector index mapping based on the Milvus schema. Pay close attention to mapping vector fields.

1. Log in to the CSS management console.
2. In the navigation pane on the left, choose Clusters > Elasticsearch.
3. In the cluster list, find the target cluster, and click Kibana in the Operation column to log in to the Kibana console.
4. In the navigation pane on the left, choose Dev Tools.
5. Run the following command to create a vector index using the information collected in Step 2: Analyze the Source Data Structure in 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"
         }
       }
     }
   }

   Table 7 lists the key parameters for creating a vector index. For more information, see Creating a Vector Index.

   Table 7 Key vector index parameters

   Parameter	Description
   index.vector	Must be set to true to enable vector indexing.
   refresh_interval	Set this parameter to -1 during bulk data import to disable automatic refreshes. Restore the original value after the import is complete.
   type	Must be set to vector for vector fields.
   dimension	Vector dimension. This value must match the dimension of the corresponding vector field in the source Milvus collection. Value range: 1–4096.
   indexing	Specifies whether to build a vector index for a specified field. Setting this parameter to true enables vector search for the field.
   algorithm	Vector indexing algorithm. Supported values include FLAT, GRAPH, and GRAPH_PQ.
   metric	Distance metric. This value must correspond to the distance metric used in the source Milvus collection.

Step 4: Perform Data Migration

Choose one of the following methods based on your environment.

Method B: PyMilvus + Elasticsearch API Script

1. Run the following command on the ECS to install the dependencies required by the migration script.

   pip install pymilvus elasticsearch numpy

2. Create the migration configuration file config.ini.

   Run the vi config.ini command on the ECS. Enter the following content, and run :wq to save the file.

   [Milvus]
   # Milvus host address
   host = <milvus_host>
   # Milvus service port
   port = <milvus_port>
   # User name
   username = '<milvus_username>'
   # Password
   password = '<milvus_password>'
   # Database name
   database = default
   # Name of the collection to migrate
   collection = <collection_name>
   # Number of documents read in each batch. Range: 100 to 1000.
   batch_size = 1000
   # Primary key field
   primary_key = id
   # Vector field
   vector_field = <vector_field_name>
   # Vector dimension. If this parameter is left blank, the value is automatically obtained from the schema.
   vector_dim = 
   # Text fields. Separate multiple fields with commas (,). Mapped to text + keyword.
   text_fields = 
   # Fields to migrate
   schema_fields = id,vector,color
   # Milvus data query expression. Leaving this parameter blank means to migrate all data.
   expr = 
   
   [Elasticsearch]
   # Elasticsearch endpoint in the form of host:port
   endpoint = https://<css_host>:9200
   # Elasticsearch cluster username, required only if security mode is enabled.
   username = <es_username>
   # Elasticsearch cluster password, required only if security mode is enabled.
   password = <es_password>
   # Destination index name. If this parameter is left blank, the Milvus collection name is used.
   index_name = 
   # Number of shards of the destination index
   number_of_shards = 3
   # Whether to export data to files
   export_file_flag = false

3. Create the migration script milvus2es.py.

   Execute vi milvus2es.py on the ECS. Edit the script by referring to Appendix: Migration Script, and save the script by entering :wq.

4. Start the migration script to migrate data from Milvus to CSS.
   • Create the CSS index, import data, and verify the result in a single operation.

     python milvus2es.py

   • Alternatively, use the --step parameter to specify the execution steps.

     # Create the CSS vector index: Creates an index in CSS based on the Milvus collection schema, including field type mappings and vector field configuration.
     python milvus2es.py --step mapping
     
     # Import data: Reads data from Milvus in batches and imports it into CSS. The script retries interrupted imports automatically.
     python milvus2es.py --step data
     
     # Verify the result: Compares the document counts in Milvus and CSS, and verifies data integrity through sample checks.
     python milvus2es.py --step verify

5. Example output of the migration script:

   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 - Successfully connected to 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 - Total Milvus collection entities: 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 - Number of Milvus documents: 490, number of Elasticsearch documents: 10
   2025-12-05 10:30:31,211 - WARNING - Inconsistent document counts. Milvus: 490, Elasticsearch: 10
   2025-12-05 10:30:31,211 - INFO - Migration completed in 0:00:01.778907

Step 5: Verify Data Integrity

After the migration is complete, verify that the data in the CSS cluster is consistent with that in the source Milvus cluster.

Log in to Kibana for the destination Elasticsearch cluster, and run the following command on the Dev Tools page to check the number of migrated documents:

1. Log in to Kibana for the destination Elasticsearch cluster.
2. In the navigation pane on the left, choose Dev Tools.
3. Run the following command to view the number of migrated documents.

   GET /milvus_migrated_collection/_count

   Replace milvus_migrated_collection with the name of the destination index.

4. Compare the document count in the CSS index with the Entity count obtained in Step 2: Analyze the Source Data Structure in Milvus.
5. (Optional) If lazy_indexing was enabled when the vector index was created, run the following command to build the vector index offline before performing a standard VectorQuery.

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

6. Run a vector search to verify it.

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

7. (Optional) If automatic index refreshes were disabled during data import, restore the refresh_interval setting after the migration has been verified.

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

FAQ

• Question 1: What should I do if vector search results in CSS are inconsistent with those in Milvus after migration?

  Possible causes and solutions:

  • Distance metric mismatch: Ensure that Milvus and CSS use the same (or equivalent) distance metrics. For the mapping, see Table 2.
  • Different HNSW parameters: Differences in HNSW parameters (such as neighbors, efc, and ef) between CSS and Milvus can affect recall performance. Tune the vector index parameters in CSS as needed.
  • Floating-point precision: Minor precision loss may occur during data export or import. To minimize this, use the double data type when storing intermediate results.
• Q2: What should I do if migrating hundreds of millions of records takes too long?
  • Migrate in batches: Divide the migration by Milvus partitions or primary key ranges to reduce the workload of individual migration tasks.
  • Run migrations in parallel: Launch multiple VTS instances or migration script processes to migrate different collections or partitions simultaneously.
  • Scale out the CSS cluster: Temporarily increase the size of the CSS cluster to accelerate data ingestion and indexing.
  • Enable deferred indexing: Set lazy_indexing to true for the CSS vector index so that the index is built only after all data has been imported.
• What should I do if new data is written to Milvus during the migration?

  Solutions:

  • Stop writes during migration: Suspend writes to Milvus during off-peak hours, then switch applications to CSS after the migration is complete.
  • Use dual writes: Configure applications to write to both Milvus and CSS during the migration, and stop writing to Milvus after the migration is complete.
  • Perform incremental synchronization: Record the migration start timestamp. After the initial migration finishes, query Milvus for data added after that timestamp and import the incremental data into CSS.

Appendix: Migration Script

When using the PyMilvus + Elasticsearch API migration method, you need to create a migration script milvus2es.py. The following is an example:

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


# Import data into 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()}")
            # Test the connection.
            if es.ping():
                logger.info(f"Connected to Elasticsearch: {self.es_host}")
            else:
                raise Exception("Failed to connect to 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"
            # Vector types are handled separately in Elasticsearch. This function maps only scalar data types.
        }
        return type_map.get(milvus_type, "text")
    def __create_es_index(self, fields_mapping: Dict[str, Any]) -> bool:
        try:
            # Check whether the index already exists.
            if self.es_client.indices.exists(index=self.es_index_name):
                logger.warning(f"Index {self.es_index_name} already exists")
                return True

            # Build the index mapping
            mapping = {
                "settings": {
                    "index": {
                        "vector": true,  # Enable vector indexing.
                        "number_of_shards": self.number_of_shards,  # Number of primary shards
                        "number_of_replicas": 1,
                        "refresh_interval": "1s"
                    }
                },
                "mappings": {  # Add the mapping section.
                    "properties": {}  # Add the properties section.
                }
            }
            for field_name, field_config in fields_mapping.items():
                mapping["mappings"]["properties"][field_name] = field_config
            # Create an index.
            self.es_client.indices.create(index=self.es_index_name, body=mapping)
            logger.info(f"Create Elasticsearch index {self.es_index_name} and mapping {mapping}")
            return True
        except Exception as e:
            logger.error(f"Failed to create Elasticsearch index: {e}")
            return False

    def create_elasticsearch_index(self) -> None:
        try:
            total_entities = self.collection.num_entities
            logger.info(f"Total entities in the Milvus collection: {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

            # Obtain the collection schema
            schema = self.collection.schema
            # Build Elasticsearch field mappings.
            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:
                    # Map scalar fields based on their data types.
                    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

            # Prepare bulk operations for Elasticsearch.
            actions = []
            for entity in results:
                doc_id = entity.get(self.primary_key)
                # Build documents.
                doc_source = {}
                for field in self.text_fields:
                    if field in entity:
                        doc_source[field] = entity[field]
                # Process vector fields.
                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
                # Add the remaining fields.
                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:
            # Compare document counts. If duplicate primary keys exist in Milvus and have been updated, the document counts will differ.
            milvus_count = self.collection.num_entities
            es_count = self.es_client.count(index=self.es_index_name)['count']
            logger.info(f"Milvus document count: {milvus_count}, Elasticsearch document count: {es_count}")
            if milvus_count != es_count:
                logger.warning(f"Document count mismatch: Milvus={milvus_count}, Elasticsearch={es_count}")
                return False
            # Perform sample-based validation.
            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"Document {doc_id} does not exist in Elasticsearch")
                        return False

            logger.info("Migration verification succeeded")
            return True
        except Exception as e:
            logger.error(f"Failed to verify migration result: {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='Specify steps to execute. Options include mapping, data, and verify. If unspecified, all steps are executed.')
    parser.add_argument('--config', type=str, default='config.ini',
                        help='Configuration file path. Default: 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)

Related Documents

• CSS Vector Database: Learn how to use the CSS vector database, including creating vector indexes, importing vector data, and performing vector search.
• Using Elasticsearch for Vector Search: Learn how to quickly get started with CSS vector search.
• VTS: Learn how to use the Vector Transport Service (VTS).
• PyMilvus SDK: Learn how to install and use the PyMilvus SDK.