集合创建与 Schema 定义

概述

在 Milvus 中，Collection（集合）是存储数据的基本容器。创建集合前，需要先定义 Schema（模式），描述集合中字段的结构和属性。

定义字段（FieldSchema）

主键字段

每个集合必须有一个主键字段：

python

from pymilvus import FieldSchema, DataType

# 使用整数作为主键
id_field = FieldSchema(
    name="id",                    # 字段名称
    dtype=DataType.INT64,         # 数据类型
    is_primary=True,              # 是否为主键
    auto_id=True                  # 是否自动递增
)

# 使用字符串作为主键
name_field = FieldSchema(
    name="name",
    dtype=DataType.VARCHAR,
    is_primary=True,
    auto_id=False,                # 字符串主键不能自增
    max_length=100                # 字符串最大长度
)

向量字段

python

from pymilvus import FieldSchema, DataType

# 浮点向量（最常用）
vector_field = FieldSchema(
    name="embedding",             # 字段名称
    dtype=DataType.FLOAT_VECTOR,  # 向量类型
    dim=128                       # 向量维度
)

# 二进制向量
binary_vector_field = FieldSchema(
    name="binary_embedding",
    dtype=DataType.BINARY_VECTOR,
    dim=128
)

标量字段

python

from pymilvus import FieldSchema, DataType

# 字符串字段
title_field = FieldSchema(
    name="title",
    dtype=DataType.VARCHAR,
    max_length=512                # 必须指定最大长度
)

# 整数字段
category_id_field = FieldSchema(
    name="category_id",
    dtype=DataType.INT64
)

# 浮点数字段
score_field = FieldSchema(
    name="score",
    dtype=DataType.FLOAT
)

# 布尔字段
is_active_field = FieldSchema(
    name="is_active",
    dtype=DataType.BOOL
)

# JSON 字段
metadata_field = FieldSchema(
    name="metadata",
    dtype=DataType.JSON
)

定义集合 Schema

基础 Schema

python

from pymilvus import CollectionSchema, FieldSchema, DataType

# 定义字段
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=128),
    FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=512),
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=64)
]

# 创建 Schema
schema = CollectionSchema(
    fields=fields,
    description="文章搜索集合",
    enable_dynamic_field=True      # 启用动态字段
)

启用动态字段

动态字段允许插入未在 Schema 中预定义的字段：

python

schema = CollectionSchema(
    fields=fields,
    description="支持动态字段的集合",
    enable_dynamic_field=True
)

# 插入时可以包含额外字段
data = {
    "vector": [0.1, 0.2, ...],
    "title": "示例文章",
    "extra_field": "这个字段不在 Schema 中"  # 动态字段
}

创建集合

基础创建

python

from pymilvus import Collection

# 创建集合
collection = Collection(
    name="article_collection",     # 集合名称
    schema=schema
)

print(f"集合 '{collection.name}' 创建成功")

完整示例

python

from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection

# 连接 Milvus
connections.connect(host="localhost", port="19530")

# 定义字段
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="article_vector", dtype=DataType.FLOAT_VECTOR, dim=768),
    FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=512),
    FieldSchema(name="content", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=64),
    FieldSchema(name="publish_time", dtype=DataType.INT64),
    FieldSchema(name="read_count", dtype=DataType.INT64),
    FieldSchema(name="tags", dtype=DataType.ARRAY, element_type=DataType.VARCHAR, max_length=32, max_capacity=10)
]

# 创建 Schema
schema = CollectionSchema(
    fields=fields,
    description="文章语义搜索集合",
    enable_dynamic_field=True
)

# 创建集合
collection = Collection(name="article_search", schema=schema)
print("集合创建成功！")

集合属性

设置集合属性

python

from pymilvus import Collection

# 创建集合时设置属性
collection = Collection(
    name="my_collection",
    schema=schema,
    properties={
        "collection.ttl.seconds": 86400,      # 数据存活时间（秒）
        "collection.autoCompaction.enabled": True  # 启用自动压缩
    }
)

常用集合属性

属性名	说明	默认值
`collection.ttl.seconds`	数据自动过期时间	-1（不过期）
`collection.autoCompaction.enabled`	自动压缩	false
`collection.maxCapacity`	最大容量	无限制

管理集合

列出所有集合

python

from pymilvus import utility

# 获取所有集合名称
collections = utility.list_collections()
print(f"所有集合: {collections}")

检查集合是否存在

python

from pymilvus import utility

exists = utility.has_collection("article_search")
print(f"集合是否存在: {exists}")

获取集合信息

python

from pymilvus import Collection

collection = Collection("article_search")

# 获取 Schema
schema = collection.schema
print(f"Schema: {schema}")

# 获取描述
description = collection.description
print(f"描述: {description}")

# 获取字段
fields = collection.schema.fields
for field in fields:
    print(f"字段: {field.name}, 类型: {field.dtype}")

删除集合

python

from pymilvus import utility

# 删除集合
utility.drop_collection("article_search")
print("集合已删除")

# 删除前确认
if utility.has_collection("article_search"):
    utility.drop_collection("article_search")

重命名集合

python

from pymilvus import utility

# 重命名集合
utility.rename_collection(
    old_collection_name="old_name",
    new_collection_name="new_name"
)

加载和释放集合

加载集合到内存

搜索前需要将集合加载到内存：

python

from pymilvus import Collection

collection = Collection("article_search")

# 加载整个集合
collection.load()

# 加载指定分区
collection.load(partition_names=["partition_2024"])

# 异步加载
collection.load(_async=True)

释放集合

python

# 释放集合
collection.release()

# 释放指定分区
collection.release(partition_names=["partition_2024"])

检查加载状态

python

# 检查集合是否已加载
is_loaded = collection.is_loaded
print(f"集合已加载: {is_loaded}")

# 获取加载进度
progress = collection.load_state()
print(f"加载状态: {progress}")

完整实战示例

python

from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility

def create_image_search_collection():
    """创建图片搜索集合"""
    
    # 连接 Milvus
    connections.connect(host="localhost", port="19530")
    
    collection_name = "image_search"
    
    # 如果集合已存在，先删除
    if utility.has_collection(collection_name):
        utility.drop_collection(collection_name)
        print(f"已删除旧集合: {collection_name}")
    
    # 定义字段
    fields = [
        # 主键字段
        FieldSchema(
            name="image_id",
            dtype=DataType.INT64,
            is_primary=True,
            auto_id=True,
            description="图片唯一标识"
        ),
        
        # 向量字段（CLIP 模型通常输出 512 维向量）
        FieldSchema(
            name="image_vector",
            dtype=DataType.FLOAT_VECTOR,
            dim=512,
            description="图片特征向量"
        ),
        
        # 标量字段
        FieldSchema(
            name="image_url",
            dtype=DataType.VARCHAR,
            max_length=1024,
            description="图片 URL"
        ),
        
        FieldSchema(
            name="image_name",
            dtype=DataType.VARCHAR,
            max_length=256,
            description="图片名称"
        ),
        
        FieldSchema(
            name="category",
            dtype=DataType.VARCHAR,
            max_length=64,
            description="图片分类"
        ),
        
        FieldSchema(
            name="upload_time",
            dtype=DataType.INT64,
            description="上传时间戳"
        ),
        
        FieldSchema(
            name="file_size",
            dtype=DataType.INT64,
            description="文件大小（字节）"
        ),
        
        FieldSchema(
            name="tags",
            dtype=DataType.ARRAY,
            element_type=DataType.VARCHAR,
            max_length=32,
            max_capacity=20,
            description="图片标签"
        ),
        
        FieldSchema(
            name="metadata",
            dtype=DataType.JSON,
            description="额外元数据"
        )
    ]
    
    # 创建 Schema
    schema = CollectionSchema(
        fields=fields,
        description="图片语义搜索集合，支持以图搜图",
        enable_dynamic_field=True
    )
    
    # 创建集合
    collection = Collection(
        name=collection_name,
        schema=schema,
        properties={
            "collection.ttl.seconds": 2592000  # 30 天过期
        }
    )
    
    print(f"集合 '{collection_name}' 创建成功！")
    print(f"字段数量: {len(collection.schema.fields)}")
    
    return collection

if __name__ == "__main__":
    collection = create_image_search_collection()

字段类型参考表

数据类型	Python 对应类型	说明
`DataType.BOOL`	bool	布尔值
`DataType.INT8`	int	8位整数
`DataType.INT16`	int	16位整数
`DataType.INT32`	int	32位整数
`DataType.INT64`	int	64位整数
`DataType.FLOAT`	float	32位浮点
`DataType.DOUBLE`	float	64位浮点
`DataType.VARCHAR`	str	变长字符串
`DataType.JSON`	dict	JSON 对象
`DataType.ARRAY`	list	数组类型
`DataType.FLOAT_VECTOR`	list[float]	浮点向量
`DataType.BINARY_VECTOR`	bytes	二进制向量

下一步

创建集合后，你需要：

集合创建与 Schema 定义 ​

概述 ​

定义字段（FieldSchema） ​

主键字段 ​

向量字段 ​

标量字段 ​

定义集合 Schema ​

基础 Schema ​

启用动态字段 ​

创建集合 ​

基础创建 ​

完整示例 ​

集合属性 ​

设置集合属性 ​

常用集合属性 ​

管理集合 ​

列出所有集合 ​

检查集合是否存在 ​

获取集合信息 ​

删除集合 ​

重命名集合 ​

加载和释放集合 ​

加载集合到内存 ​

释放集合 ​

检查加载状态 ​

完整实战示例 ​

字段类型参考表 ​

下一步 ​

集合创建与 Schema 定义

概述

定义字段（FieldSchema）

主键字段

向量字段

标量字段

定义集合 Schema

基础 Schema

启用动态字段

创建集合

基础创建

完整示例

集合属性

设置集合属性

常用集合属性

管理集合

列出所有集合

检查集合是否存在

获取集合信息

删除集合

重命名集合

加载和释放集合

加载集合到内存

释放集合

检查加载状态

完整实战示例

字段类型参考表

下一步