当前位置：首页 > news >正文

Milvus：集合（Collections）操作详解（三）

news 2025/11/4 16:52:50

一、Collections 核心概念

在开始实操前，需先理解 Collections 及其关联概念，这是后续操作的基础。

1.1 什么是 Collections？

Collections 是 Milvus 中用于管理数据的 “二维表”，类比关系数据库中的 “表”，主要特点如下：

结构固定：每一列对应一个 “字段（Field）”，字段的类型、约束等需提前定义。
数据灵活：每一行对应一个 “实体（Entity）”，实体数量可根据需求动态增减。
核心作用：所有数据（向量、标量等）需插入到 Collections 中，才能进行后续的搜索和查询。

1.2 关键关联概念

二、Collections 核心操作

本节是重点，将按 “创建 → 查看 → 修改 → 加载 / 释放 → 删除” 的流程，提供完整代码示例。

2.1 创建 Collections（核心步骤）

创建 Collections 需先定义 Schema（数据结构），再设置可选的 索引参数 和 集合属性（如 TTL、分片数）。

步骤 1：初始化 Milvus 客户端

from pymilvus import MilvusClient, DataType# 连接 Milvus 服务
client = MilvusClient(uri="http://localhost:19530",  # 服务地址token="root:Milvus"            # 认证 Token（默认值）
)

步骤 2：定义 Schema（数据结构）

Schema 需包含主键字段和向量字段（必选），可添加标量字段（如 VARCHAR）和动态字段（可选）。

# 1. 创建 Schema 对象（关闭 AutoId，开启动态字段）
schema = client.create_schema(auto_id=False,               # 关闭自动生成主键（需手动传入主键值）enable_dynamic_field=True    # 开启动态字段（支持插入 Schema 未定义的字段）
)# 2. 向 Schema 中添加字段
# 字段 1：主键（INT64 类型，唯一标识实体）
schema.add_field(field_name="user_id",        # 字段名datatype=DataType.INT64,     # 数据类型is_primary=True              # 标记为主键
)# 字段 2：向量字段（FLOAT_VECTOR 类型，维度为 5）
schema.add_field(field_name="user_embedding", # 字段名datatype=DataType.FLOAT_VECTOR,dim=5                        # 向量维度（需与实际数据一致）
)# 字段 3：标量字段（VARCHAR 类型，存储用户名称）
schema.add_field(field_name="user_name",datatype=DataType.VARCHAR,max_length=512               # VARCHAR 类型必须指定最大长度
)

步骤 3：设置索引参数（可选）

索引用于提升搜索效率，向量字段必须创建索引，标量字段可根据过滤需求创建。

# 1. 初始化索引参数对象
index_params = client.prepare_index_params()# 2. 为主键添加索引（AUTOINDEX 自动选择合适的索引类型）
index_params.add_index(field_name="user_id",index_type="AUTOINDEX"
)# 3. 为向量字段添加索引（指定度量类型为 COSINE，用于计算向量相似度）
index_params.add_index(field_name="user_embedding",index_type="AUTOINDEX",metric_type="COSINE"  # 度量类型：COSINE（余弦相似度）、L2（欧氏距离）、IP（内积）
)

步骤 4：创建 Collections

可选择 “创建时自动加载”（带索引参数）或 “创建后手动加载”（不带索引参数）。

# 方式 1：创建时带索引参数（自动加载 Collections）
client.create_collection(collection_name="user_collection",  # 集合名称（自定义）schema=schema,                      # 传入定义好的 Schemaindex_params=index_params,          # 传入索引参数num_shards=1,                       # 分片数（默认 1，数据量大时可增加）properties={"collection.ttl.seconds": 86400}  # TTL：数据保留 1 天（86400 秒）
)# 检查加载状态（应返回 "Loaded"）
load_state = client.get_load_state(collection_name="user_collection")
print("加载状态：", load_state)  # 输出：{"state": "<LoadState: Loaded>"}# 方式 2：创建时不带索引参数（后续手动添加索引，初始状态为 "NotLoad"）
# client.create_collection(
#     collection_name="user_collection_no_index",
#     schema=schema
# )

2.2 查看 Collections 信息

创建后可通过代码查看集合列表、集合详情等信息。

2.2.1 列出所有 Collections

# 获取当前数据库中所有集合名称
collections = client.list_collections()
print("所有集合：", collections)  # 输出：["user_collection"]

2.2.2 查看集合详情（Schema、索引、属性等）

# 查看 "user_collection" 的详细信息
collection_info = client.describe_collection(collection_name="user_collection")
print("集合详情：", collection_info)# 关键输出解读：
# - "fields"：显示所有字段的定义（类型、是否主键等）
# - "num_shards"：分片数
# - "properties"：TTL 等属性
# - "consistency_level"：一致性级别（默认 Bounded）

2.3 修改 Collections

支持修改集合属性（如 TTL、动态字段）、重命名集合、管理别名等操作。

2.3.1 修改集合属性（以 TTL 为例）

# 1. 更改 TTL 为 2 天（172800 秒）
client.alter_collection_properties(collection_name="user_collection",properties={"collection.ttl.seconds": 172800}
)# 2. 删除 TTL（数据永久保留）
# client.drop_collection_properties(
#     collection_name="user_collection",
#     property_keys=["collection.ttl.seconds"]
# )

2.3.2 重命名集合

# 将 "user_collection" 重命名为 "user_collection_v2"
client.rename_collection(old_name="user_collection",new_name="user_collection_v2"
)

2.3.3 管理别名（生产环境常用）

别名可动态指向不同集合，适合无缝切换数据（如 A/B 测试、数据更新）。

# 1. 为集合创建别名（一个集合可多个别名）
client.create_alias(collection_name="user_collection_v2",alias="user_prod"  # 生产环境用的别名
)# 2. 列出集合的所有别名
aliases = client.list_aliases(collection_name="user_collection_v2")
print("集合别名：", aliases)  # 输出：{"aliases": ["user_prod"], ...}# 3. 切换别名（指向新集合）
# client.create_collection(collection_name="user_collection_v3", schema=schema)  # 新建集合
# client.alter_alias(collection_name="user_collection_v3", alias="user_prod")  # 别名指向新集合# 4. 删除别名
# client.drop_alias(alias="user_prod")

2.4 加载与释放 Collections

Milvus 搜索 / 查询需先将集合加载到内存，不使用时释放内存以节约资源。

2.4.1 加载集合（全量加载或部分字段加载）

# 方式 1：全量加载（加载所有字段）
client.load_collection(collection_name="user_collection_v2")# 方式 2：部分字段加载（仅加载需要的字段，减少内存占用）
# client.load_collection(
#     collection_name="user_collection_v2",
#     load_fields=["user_id", "user_embedding"],  # 仅加载主键和向量字段
#     skip_load_dynamic_field=True  # 不加载动态字段
# )# 检查加载状态
load_state = client.get_load_state(collection_name="user_collection_v2")
print("加载状态：", load_state)  # 输出：{"state": "<LoadState: Loaded>"}

2.4.2 释放集合（不使用时释放内存）

# 释放集合（内存中的数据会被清除，不影响磁盘存储）
client.release_collection(collection_name="user_collection_v2")# 检查释放状态
load_state = client.get_load_state(collection_name="user_collection_v2")
print("释放后状态：", load_state)  # 输出：{"state": "<LoadState: NotLoad>"}

2.5 管理分区

分区是集合的子集，可按业务逻辑（如时间、地域）划分数据，提升搜索效率。

2.5.1 创建分区

# 为 "user_collection_v2" 创建分区（按用户地域划分）
client.create_partition(collection_name="user_collection_v2",partition_name="user_partition_beijing"  # 分区名称
)# 列出集合的所有分区（默认有 "_default" 分区）
partitions = client.list_partitions(collection_name="user_collection_v2")
print("所有分区：", partitions)  # 输出：["_default", "user_partition_beijing"]

2.5.2 加载 / 释放分区

# 加载指定分区
client.load_partitions(collection_name="user_collection_v2",partition_names=["user_partition_beijing"]
)# 释放指定分区
# client.release_partitions(
#     collection_name="user_collection_v2",
#     partition_names=["user_partition_beijing"]
# )

2.5.3 删除分区（需先释放）

# 1. 先释放分区
client.release_partitions(collection_name="user_collection_v2",partition_names=["user_partition_beijing"]
)# 2. 删除分区（分区内数据会被永久删除）
client.drop_partition(collection_name="user_collection_v2",partition_name="user_partition_beijing"
)

2.6 删除 Collections

不再需要的集合可永久删除（需谨慎，删除后数据无法恢复）。

# 删除 "user_collection_v2"（删除前无需释放，但建议先释放内存）
client.drop_collection(collection_name="user_collection_v2")# 验证是否删除（列表中应无该集合）
collections = client.list_collections()
print("删除后所有集合：", collections)  # 输出：[]