从Milvus迁移DashVector

2024-09-05 17:12
文章标签 milvus 迁移 dashvector

本文主要是介绍从Milvus迁移DashVector,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

本文档演示如何从Milvus将Collection数据全量导出,并适配迁移至DashVector。方案的主要流程包括:

  1. 首先,升级Milvus版本,目前Milvus只有在最新版本(v.2.3.x)中支持全量导出
  2. 其次,将Milvus Collection的Schema信息和数据信息导出到具体的文件中
  3. 最后,以导出的文件作为输入来构建DashVector Collection并数据导入

下面,将详细阐述迁移方案的具体操作细节。

1. Milvus升级2.3.x版本

本文中,我们将借助Milvus的query_iterator来全量导出数据(query接口无法导出完整数据),由于该接口目前只在v2.3.x版本中支持,所以在导出数据前,需要先将Milvus版本升级到该版本。Milvus版本升级的详细操作参考Milvus用户文档。

注意:在进行Milvus Upgrade时需要注意数据的备份安全问题。

2. Milvus全量数据导出

数据的导出包含Schema以及数据记录,Schema主要用于完备地定义Collection,数据记录对应于每个Partition下的全量数据,这两部分涵盖了需要导出的全部数据。下文展示如何将单个Milvus Collection全量导出。

2.1. Schema导出

DashVector和Milvus在Schema的设计上有一些区别,DashVector向用户透出的接口非常简单,Milvus则更加详尽。从Milvus迁移DashVector时会涉及到部分Schema参数的删除(例如Collection的index_param参数),只会保留DashVector构建Collection的必要参数,以下为一个Schema转换的简单示例(其中,Collection已有的数据参考Milvus示例代码写入)。

python示例:

from pymilvus import (connections,utility,Collection,DataType
)
import os
import json
from pathlib import Pathfmt = "\n=== {:30} ===\n"print(fmt.format("start connecting to Milvus"))
host = os.environ.get('MILVUS_HOST', "localhost")
print(fmt.format(f"Milvus host: {host}"))
connections.connect("default", host=host, port="19530")metrics_map = {'COSINE': 'cosine','L2': 'euclidean','IP': 'dotproduct',
}dtype_map = {DataType.BOOL: 'bool',DataType.INT8: 'int',DataType.INT16: 'int',DataType.INT32: 'int',DataType.INT64: 'int',DataType.FLOAT: 'float',DataType.DOUBLE: 'float',DataType.STRING: 'str',DataType.VARCHAR: 'str',
}def load_collection(collection_name: str) -> Collection:has = utility.has_collection(collection_name)print(f"Does collection hello_milvus exist in Milvus: {has}")if not has:return Nonecollection = Collection(collection_name)      collection.load()return collectiondef export_collection_schema(collection, file: str):schema = collection.schema.to_dict()index = collection.indexes[0].to_dict()export_schema = dict()milvus_metric_type = index['index_param']['metric_type']try:export_schema['metrics'] = metrics_map[milvus_metric_type]except:raise Exception(f"milvus metrics_type{milvus_metric_type} not supported")export_schema['fields_schema'] = {}for field in schema['fields']:if 'is_primary' in field and field['is_primary']:continueif field['name'] == index['field']:# vectorif field['type'] == DataType.FLOAT_VECTOR:export_schema['dtype'] = 'float'export_schema['dimension'] = field['params']['dim']else:raise Exception(f"milvus dtype{field['type']} not supported yet")else:try:# non-vectorexport_schema['fields_schema'][field['name']] = dtype_map[field['type']]except:raise Exception(f"milvus dtype{field['type']} not supported yet")with open(file, 'w') as file:json.dump(export_schema, file, indent=4)  if __name__ == "__main__":collection_name = "YOUR_MILVUS_COLLECTION_NAME"collection = load_collection(collection_name)dump_path_str = collection_name+'.dump'dump_path = Path(dump_path_str)dump_path.mkdir(parents=True, exist_ok=True)schema_file = dump_path_str + "/schema.json"export_collection_schema(collection, schema_file)

JSON示例:

{"metrics": "euclidean","fields_schema": {"random": "float","var": "str"},"dtype": "float","dimension": 8
}

2.2. Data导出

DashVector和Milvus在设计上都有Partition的概念,所以向量以及其他数据进行导出时,需要注意按照Partition粒度进行导出。此外,DashVector的主键类型为str,而Milvus设计其为自定义类型,所以在导出时需要考虑主键类型的转换。以下为一个基于query_iterator接口导出的简单代码示例:

from pymilvus import (connections,utility,Collection,DataType
)
import os
import json
import numpy as np
from pathlib import Pathfmt = "\n=== {:30} ===\n"print(fmt.format("start connecting to Milvus"))
host = os.environ.get('MILVUS_HOST', "localhost")
print(fmt.format(f"Milvus host: {host}"))
connections.connect("default", host=host, port="19530")
pk = "pk"
vector_field_name = "vector"def load_collection(collection_name: str) -> Collection:has = utility.has_collection(collection_name)print(f"Does collection hello_milvus exist in Milvus: {has}")if not has:return Nonecollection = Collection(collection_name)      collection.load()return collectiondef export_partition_data(collection, partition_name, file: str):batch_size = 10output_fields=["pk", "random", "var", "embeddings"]query_iter = collection.query_iterator(batch_size=batch_size,output_fields = output_fields,partition_names=[partition_name])export_file = open(file, 'w')while True:docs = query_iter.next()if len(docs) == 0:# close the iteratorquery_iter.close()breakfor doc in docs:new_doc = {}new_doc_fields = {}for k, v in doc.items():if k == pk:# primary keynew_doc['pk'] = str(v)elif k == vector_field_name:new_doc['vector'] = [float(k) for k in v]else:new_doc_fields[k] = vnew_doc['fields'] = new_doc_fieldsjson.dump(new_doc, export_file)export_file.write('\n')export_file.close()if __name__ == "__main__":collection_name = "YOUR_MILVUS_COLLECTION_NAME"collection = load_collection(collection_name)pk = collection.schema.primary_field.namevector_field_name = collection.indexes[0].field_namedump_path_str = collection_name+'.dump'dump_path = Path(dump_path_str)dump_path.mkdir(parents=True, exist_ok=True)for partition in collection.partitions:partition_name = partition.nameif partition_name == '_default':export_path = dump_path_str + '/default.txt'else:export_path = dump_path_str + '/' + partition_name + ".txt"export_partition_data(collection, partition_name, export_path)

3. 将数据导入DashVector

3.1. 创建Cluster

参考DashVector官方用户手册构建Cluster。

3.2. 创建Collection

根据2.1章节中导出的Schema信息以及参考Dashvector官方用户手册来创建Collection。下面的示例代码会根据2.1章节中导出的schema.json来创建一个DashVector的Collection。

from dashvector import Client, DashVectorExceptionfrom pydantic import BaseModel
from typing import Dict, Type
import jsondtype_convert = {'int': int,'float': float,'bool': bool,'str': str
}class Schema(BaseModel):metrics: strdtype: Typedimension: intfields_schema: Dict[str, Type]@classmethoddef from_dict(cls, json_data):metrics = json_data['metrics']dtype = dtype_convert[json_data['dtype']]dimension = json_data['dimension']fields_schema = {k: dtype_convert[v] for k, v in json_data['fields_schema'].items()}return cls(metrics=metrics, dtype=dtype, dimension=dimension, fields_schema=fields_schema)def read_schema(schema_path) -> Schema:with open(schema_path) as file:json_data = json.loads(file.read())return Schema.from_dict(json_data)if __name__ == "__main__":milvus_dump_path = f"{YOUR_MILVUS_COLLECTION_NAME}.dump"milvus_dump_scheme_path = milvus_dump_path + "/schema.json"schema = read_schema(milvus_dump_scheme_path)client = dashvector.Client(api_key='YOUR_API_KEY',endpoint='YOUR_CLUSTER_ENDPOINT')# create collectionrsp = client.create(name="YOUR_DASHVECTOR_COLLECTION_NAME", dimension=schema.dimension, metric=schema.metrics, dtype=schema.dtype,fields_schema=schema.fields_schema)if not rsp:raise DashVectorException(rsp.code, reason=rsp.message)

3.3. 导入Data

根据2.2章节中导出的数据以及参考DashVector官方用户手册来批量插入Doc。下面的示例代码会依次解析各个Partition导出的数据,然后依次创建DashVector下的Partition并导入数据。

from dashvector import Client, DashVectorException, Docfrom pydantic import BaseModel
from typing import Dict, Type
import json
import glob
from pathlib import Pathdef insert_data(collection, partition_name, partition_file):if partition_name != 'default':rsp = collection.create_partition(partition_name)if not rsp:raise DashVectorException(rsp.code, reason=rsp.message)with open(partition_file) as f:for line in f:if line.strip():json_data = json.loads(line)rsp = collection.insert([Doc(id=json_data['pk'], vector=json_data['vector'], fields=json_data['fields'])])if not rsp:raise DashVectorException(rsp.code, reason=rsp.message)  if __name__ == "__main__":milvus_dump_path = f"{YOUR_MILVUS_COLLECTION_NAME}.dump"client = dashvector.Client(api_key='YOUR_API_KEY',endpoint='YOUR_CLUSTER_ENDPOINT')# create collectioncollection = client.get("YOUR_DASHVECTOR_COLLECTION_NAME")partition_files = glob.glob(milvus_dump_path+'/*.txt', recursive=False)for partition_file in partition_files:# create partitionpartition_name = Path(partition_file).steminsert_data(collection, partition_name, partition_file)

这篇关于从Milvus迁移DashVector的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1139566

相关文章

CentOs7上Mysql快速迁移脚本

因公司业务需要,对原来在/usr/local/mysql/data目录下的数据迁移到/data/local/mysql/mysqlData。 原因是系统盘太小,只有20G,几下就快满了。 参考过几篇文章,基于大神们的思路,我封装成了.sh脚本。 步骤如下: 1) 先修改好/etc/my.cnf,        ##[mysqld]       ##datadir=/data/loc

CentOS下mysql数据库data目录迁移

https://my.oschina.net/u/873762/blog/180388        公司新上线一个资讯网站,独立主机,raid5,lamp架构。由于资讯网是面向小行业,初步估计一两年内访问量压力不大,故,在做服务器系统搭建的时候,只是简单分出一个独立的data区作为数据库和网站程序的专区,其他按照linux的默认分区。apache,mysql,php均使用yum安装(也尝试

Linux Centos 迁移Mysql 数据位置

转自:http://www.tuicool.com/articles/zmqIn2 由于业务量增加导致安装在系统盘(20G)磁盘空间被占满了, 现在进行数据库的迁移. Mysql 是通过 yum 安装的. Centos6.5Mysql5.1 yum 安装的 mysql 服务 查看 mysql 的安装路径 执行查询 SQL show variables like

风格控制水平创新高!南理工InstantX小红书发布CSGO:简单高效的端到端风格迁移框架

论文链接:https://arxiv.org/pdf/2408.16766 项目链接:https://csgo-gen.github.io/ 亮点直击 构建了一个专门用于风格迁移的数据集设计了一个简单但有效的端到端训练的风格迁移框架CSGO框架,以验证这个大规模数据集在风格迁移中的有益效果。引入了内容对齐评分(Content Alignment Score,简称CAS)来评估风格迁移

DataGrip数据迁移

第一步 第二步  第三步  第四步 选择你刚刚到处的文件即可

从VC6迁移至VS2005 ,VS2008

最近开发平台由VC6.0升级至VS2005,需要将原有的项目迁移,特将碰到的问题归纳如下: 1消息映射 VS2005对消息的检查更为严格,以前在VC6下完全正常运行的消息映射在VS2005下编译不通过 a. ON_MESSAGE(message,OnMyMessage);   OnMyMessage返回值必须为LRESULT,其形式为:afx_msg LRESULT OnMyMessage(

基于人工智能的图像风格迁移系统

目录 引言项目背景环境准备 硬件要求软件安装与配置系统设计 系统架构关键技术代码示例 数据预处理模型训练模型预测应用场景结论 1. 引言 图像风格迁移是一种计算机视觉技术,它可以将一种图像的风格(如梵高的绘画风格)迁移到另一幅图像上,从而生成一幅具有特定艺术风格的图像。基于深度学习的图像风格迁移技术已经广泛应用于艺术创作、图像处理等领域。本文将介绍如何构建一个基于人工智能的图像风格迁移

gs_dump和gs_dumpall 迁移数据库

目录 0、源端实例收集AWR1、创建目录2、gs_dump - 业务停机3、gs_dumpall - 业务停机4、拷贝文件5、目标实例导入数据 0、源端实例收集AWR https://blog.csdn.net/hezuijiudexiaobai/article/details/134220949 1、创建目录 mkdir -p /pgdata/data/opengauss-

ASM 10G 基于RMAN 迁移

ASM 10G 基于RMAN 迁移 场景 单节点基于10G R2 的数据库,其数据文件及日志文件均存放在ASM 里,现在为业务需求,将此数据库做迁 移,迁移到另个机房,但是两个机房的网络是通畅的,为尽量减少数据的丢失及平稳迁移和经济实惠,迁 移时,数据库需停应用 工具 本次采用RMAN 的duplicate 命令来进行迁移,运用此命令简化复杂度; 一、源库和目标库的

替换Windows AD时,网络准入场景如何迁移对接国产身份域管?

Windows AD是迄今为止身份管理和访问控制领域的最佳实践,全球约90%的中大型企业采用AD作为底层数字身份基础设施,管理组织、用户、应用、网络、终端等IT资源。但随着信创建设在党政机关、金融、央国企、电力等各行各业铺开,对Windows AD域的替换成为企业信息安全建设中不可避免的议题之一。 鉴于AD在企业中的应用程度不同,可将企业分为轻度、中度及深度三类Windows AD