集合操作 #

一、Collection概述 #

1.1 什么是Collection #

Collection 是 Milvus 中存储实体的逻辑单元，类似于关系数据库中的表。

text 
Collection结构：

┌─────────────────────────────────────────┐
│         Collection: products            │
├─────────────────────────────────────────┤
│  Schema定义：                            │
│  ├── 字段列表                            │
│  ├── 描述信息                            │
│  └── 动态字段设置                        │
├─────────────────────────────────────────┤
│  数据存储：                              │
│  ├── 分区列表                            │
│  ├── 段列表                              │
│  └── 索引列表                            │
├─────────────────────────────────────────┤
│  统计信息：                              │
│  ├── 实体数量                            │
│  └── 存储大小                            │
└─────────────────────────────────────────┘

二、创建Collection #

2.1 定义Schema #

python 
from pymilvus import FieldSchema, CollectionSchema, DataType

fields = [
    FieldSchema(
        name="id",
        dtype=DataType.INT64,
        is_primary=True,
        auto_id=False,
        description="主键ID"
    ),
    FieldSchema(
        name="title",
        dtype=DataType.VARCHAR,
        max_length=512,
        description="标题"
    ),
    FieldSchema(
        name="content",
        dtype=DataType.VARCHAR,
        max_length=65535,
        description="内容"
    ),
    FieldSchema(
        name="embedding",
        dtype=DataType.FLOAT_VECTOR,
        dim=768,
        description="嵌入向量"
    )
]

schema = CollectionSchema(
    fields=fields,
    description="文档集合",
    enable_dynamic_field=True
)

2.2 创建Collection #

python 
from pymilvus import Collection

collection = Collection(
    name="documents",
    schema=schema,
    using="default",
    shards_num=2
)

2.3 创建参数说明 #

参数 类型 说明
name str Collection名称
schema CollectionSchema Schema定义
using str 连接别名
shards_num int 分片数量
num_partitions int 分区数量限制
consistency_level str 一致性级别

2.4 一致性级别 #

python 
from pymilvus import ConsistencyLevel

collection = Collection(
    name="documents",
    schema=schema,
    consistency_level=ConsistencyLevel.STRONG
)
级别 说明 适用场景
STRONG 强一致性 金融、交易
EVENTUALLY 最终一致性 日志、分析
BOUNDED 有界一致性 通用场景
SESSION 会话一致性 用户会话

2.5 使用已有Collection #

python 
collection = Collection("documents")

print(collection.name)
print(collection.schema)
print(collection.description)

三、查看Collection #

3.1 列出所有Collection #

python 
from pymilvus import utility

collections = utility.list_collections()
print(collections)

3.2 检查Collection是否存在 #

python 
has_collection = utility.has_collection("documents")
print(has_collection)

3.3 获取Collection信息 #

python 
collection = Collection("documents")

print(collection.schema)

print(collection.description)

print(collection.num_entities)

print(collection.primary_field)

print(collection.partitions)

print(collection.indexes)

3.4 获取统计信息 #

python 
stats = collection.stats
print(stats)

四、加载与释放 #

4.1 加载Collection #

python 
collection.load()

collection.load(
    replica_number=2,
    timeout=60
)

print(collection.is_loaded)

print(collection.replica_number)

4.2 释放Collection #

python 
collection.release()

print(collection.is_loaded)

4.3 刷新数据 #

python 
collection.flush()

collection.flush(timeout=120)

4.4 压缩数据 #

python 
collection.compact()

collection.compact(timeout=300)

五、修改Collection #

5.1 设置属性 #

python 
collection.set_properties({
    "collection.ttl.seconds": 86400
})

5.2 修改一致性级别 #

python 
from pymilvus import ConsistencyLevel

collection.set_properties({
    "consistency.level": ConsistencyLevel.EVENTUALLY
})

5.3 添加字段（动态字段） #

python 
collection.set_properties({
    "collection.enable_dynamic_field": True
})

六、删除Collection #

6.1 删除Collection #

python 
from pymilvus import utility

utility.drop_collection("documents")

6.2 安全删除 #

python 
if utility.has_collection("documents"):
    utility.drop_collection("documents")

6.3 删除前检查 #

python 
collection = Collection("documents")

if collection.num_entities > 0:
    print(f"警告: Collection包含 {collection.num_entities} 条数据")

utility.drop_collection("documents")

七、Collection别名 #

7.1 创建别名 #

python 
utility.create_alias("documents", "docs")

utility.create_alias("documents", "documents_v1")

7.2 列出别名 #

python 
aliases = utility.list_aliases("documents")
print(aliases)

7.3 删除别名 #

python 
utility.drop_alias("docs")

7.4 别名应用场景 #

text 
别名使用场景：

┌─────────────────────────────────────────┐
│         版本切换                         │
├─────────────────────────────────────────┤
│                                         │
│  documents_v1 ──┐                       │
│  documents_v2 ──┼──▶ production        │
│  documents_v3 ──┘                       │
│                                         │
│  应用始终使用 "production" 别名          │
│  切换版本只需修改别名指向                │
│                                         │
└─────────────────────────────────────────┘

八、Collection最佳实践 #

8.1 Schema设计建议 #

python 
fields = [
    FieldSchema(
        name="id",
        dtype=DataType.INT64,
        is_primary=True,
        auto_id=False
    ),
    FieldSchema(
        name="embedding",
        dtype=DataType.FLOAT_VECTOR,
        dim=768
    ),
    FieldSchema(
        name="created_at",
        dtype=DataType.INT64
    ),
    FieldSchema(
        name="metadata",
        dtype=DataType.JSON
    )
]

schema = CollectionSchema(
    fields=fields,
    description="最佳实践示例",
    enable_dynamic_field=True
)

8.2 分片数量选择 #

text 
分片数量建议：

数据量          分片数
────────────────────────
< 100万         1
100万-1000万    2-4
1000万-1亿      4-8
> 1亿           8-16

考虑因素：
- 写入并发量
- 数据分布
- 查询模式

8.3 命名规范 #

python 
collection_names = [
    "user_profiles",
    "product_embeddings",
    "document_vectors",
    "image_features"
]

bad_names = [
    "UserProfiles",
    "product-embeddings",
    "document vectors",
    "123collection"
]

九、完整示例 #

9.1 创建文档Collection #

python 
from pymilvus import (
    connections,
    FieldSchema,
    CollectionSchema,
    Collection,
    DataType,
    utility
)

connections.connect("default", host="localhost", port="19530")

if utility.has_collection("documents"):
    utility.drop_collection("documents")

fields = [
    FieldSchema(
        name="doc_id",
        dtype=DataType.INT64,
        is_primary=True,
        auto_id=False
    ),
    FieldSchema(
        name="title",
        dtype=DataType.VARCHAR,
        max_length=512
    ),
    FieldSchema(
        name="content",
        dtype=DataType.VARCHAR,
        max_length=65535
    ),
    FieldSchema(
        name="author",
        dtype=DataType.VARCHAR,
        max_length=128
    ),
    FieldSchema(
        name="tags",
        dtype=DataType.ARRAY,
        element_type=DataType.VARCHAR,
        max_capacity=20,
        max_length=64
    ),
    FieldSchema(
        name="embedding",
        dtype=DataType.FLOAT_VECTOR,
        dim=768
    ),
    FieldSchema(
        name="created_at",
        dtype=DataType.INT64
    )
]

schema = CollectionSchema(
    fields=fields,
    description="文档向量集合",
    enable_dynamic_field=True
)

collection = Collection(
    name="documents",
    schema=schema,
    shards_num=2
)

print(f"Collection '{collection.name}' 创建成功")
print(f"字段数量: {len(collection.schema.fields)}")

9.2 创建图像Collection #

python 
fields = [
    FieldSchema(
        name="image_id",
        dtype=DataType.INT64,
        is_primary=True,
        auto_id=True
    ),
    FieldSchema(
        name="image_path",
        dtype=DataType.VARCHAR,
        max_length=1024
    ),
    FieldSchema(
        name="category",
        dtype=DataType.VARCHAR,
        max_length=64
    ),
    FieldSchema(
        name="width",
        dtype=DataType.INT32
    ),
    FieldSchema(
        name="height",
        dtype=DataType.INT32
    ),
    FieldSchema(
        name="feature_vector",
        dtype=DataType.FLOAT_VECTOR,
        dim=512
    )
]

schema = CollectionSchema(
    fields=fields,
    description="图像特征集合"
)

collection = Collection(
    name="images",
    schema=schema
)

十、常见问题 #

10.1 Collection已存在 #

python 
if utility.has_collection("documents"):
    print("Collection已存在")
    collection = Collection("documents")
else:
    collection = Collection("documents", schema=schema)

10.2 Schema不匹配 #

python 
existing_collection = Collection("documents")
existing_schema = existing_collection.schema

for field in schema.fields:
    if field.name not in [f.name for f in existing_schema.fields]:
        print(f"字段 {field.name} 不存在")

10.3 加载超时 #

python 
try:
    collection.load(timeout=300)
except Exception as e:
    print(f"加载超时: {e}")
    collection.load(replica_number=1)

十一、总结 #

Collection操作速查表：

操作 方法
创建 Collection(name, schema)
查看 utility.list_collections()
检查存在 utility.has_collection(name)
加载 collection.load()
释放 collection.release()
删除 utility.drop_collection(name)
创建别名 utility.create_alias()
删除别名 utility.drop_alias()

下一步，让我们学习Schema设计！

最后更新：2026-04-04