Milvus教程|麻瓜教程

Appearance

数据插入与管理 ​

数据准备 ​

单条数据插入 ​

python
from pymilvus import Collection

# 获取集合
collection = Collection("article_search")

# 准备单条数据
data = {
    "article_vector": [0.1] * 128,  # 128维向量（实际由 Embedding 模型生成）
    "title": "Milvus 向量数据库入门教程",
    "content": "Milvus 是一个开源的向量数据库...",
    "category": "技术",
    "publish_time": 1704067200,
    "read_count": 1000
}

# 插入数据
insert_result = collection.insert(data)
print(f"插入成功，ID: {insert_result.primary_keys}")

批量数据插入 ​

python
import random

# 准备批量数据
entities = []
for i in range(1000):
    entity = {
        "article_vector": [random.random() for _ in range(128)],
        "title": f"文章标题_{i}",
        "content": f"文章内容_{i}",
        "category": random.choice(["技术", "生活", "娱乐", "新闻"]),
        "publish_time": 1704067200 + i * 3600,
        "read_count": random.randint(100, 10000)
    }
    entities.append(entity)

# 批量插入
insert_result = collection.insert(entities)
print(f"成功插入 {len(insert_result.primary_keys)} 条数据")

按字段批量插入 ​

python
import random

# 准备数据（按字段组织）
article_vectors = [[random.random() for _ in range(128)] for _ in range(1000)]
titles = [f"文章标题_{i}" for i in range(1000)]
contents = [f"文章内容_{i}" for i in range(1000)]
categories = [random.choice(["技术", "生活", "娱乐"]) for _ in range(1000)]
publish_times = [1704067200 + i * 3600 for i in range(1000)]
read_counts = [random.randint(100, 10000) for _ in range(1000)]

# 插入数据
insert_result = collection.insert([
    article_vectors,
    titles,
    contents,
    categories,
    publish_times,
    read_counts
])

分区插入 ​

插入到指定分区 ​

python
# 插入到默认分区
collection.insert(data)

# 插入到指定分区
collection.insert(data, partition_name="tech_articles")

# 批量插入到分区
collection.insert(entities, partition_name="tech_articles")

数据查询 ​

基本查询 ​

python
# 查询所有数据（限制返回数量）
results = collection.query(
    expr="",  # 空表达式表示查询所有
    output_fields=["id", "title", "category"],
    limit=10
)

for result in results:
    print(f"ID: {result['id']}, 标题: {result['title']}")

条件查询 ​

python
# 根据 ID 查询
results = collection.query(
    expr="id in [1, 2, 3, 4, 5]",
    output_fields=["id", "title", "category", "read_count"]
)

# 根据分类查询
results = collection.query(
    expr='category == "技术"',
    output_fields=["id", "title", "read_count"],
    limit=20
)

# 范围查询
results = collection.query(
    expr="read_count > 5000 and read_count < 8000",
    output_fields=["id", "title", "read_count"]
)

# 时间范围查询
results = collection.query(
    expr="publish_time >= 1704067200 and publish_time <= 1706745600",
    output_fields=["id", "title", "publish_time"]
)

表达式语法 ​

操作符说明示例
==等于category == "技术"
!=不等于category != "广告"
>大于read_count > 1000
<小于read_count < 5000
>=大于等于publish_time >= 1704067200
<=小于等于publish_time <= 1706745600
in在列表中id in [1, 2, 3]
not in不在列表中id not in [4, 5, 6]
and逻辑与read_count > 1000 and category == "技术"
or逻辑或category == "技术" or category == "科学"

数据删除 ​

根据 ID 删除 ​

python
# 删除单条数据
collection.delete("id in [1]")

# 删除多条数据
collection.delete("id in [1, 2, 3, 4, 5]")

条件删除 ​

python
# 根据条件删除
collection.delete('category == "测试"')

# 复杂条件删除
collection.delete('read_count < 100 and category == "草稿"')

# 删除过期数据
collection.delete("publish_time < 1704067200")

清空集合 ​

python
# 删除所有数据（保留集合结构）
collection.delete("id >= 0")

# 或者重新创建集合
from pymilvus import utility
utility.drop_collection("article_search")
# 然后重新创建...

数据更新 ​

Milvus 不支持直接更新操作，需要通过删除后重新插入实现：

python
def update_entity(collection, entity_id, new_data):
    """更新实体数据"""
    # 1. 删除旧数据
    collection.delete(f"id in [{entity_id}]")

    # 2. 准备新数据（保留原有 ID）
    new_data["id"] = entity_id

    # 3. 插入新数据
    collection.insert(new_data)

    print(f"实体 {entity_id} 更新成功")

# 使用示例
new_data = {
    "article_vector": [0.5, 0.6, 0.7, ...],
    "title": "更新后的标题",
    "category": "技术",
    "read_count": 2000
}
update_entity(collection, 1, new_data)

Upsert 操作（Milvus 2.4+） ​

Upsert = "insert or update"，即按主键判断：主键已存在则更新，不存在则插入。它替代了传统的"先 delete 再 insert"模式，更高效且原子性更好。

Collection API 方式 ​

python
from pymilvus import Collection
import random

collection = Collection("article_search")

# 单条 upsert（若 id=1 已存在则更新，否则插入）
collection.upsert({
    "id": 1,
    "article_vector": [random.random() for _ in range(128)],
    "title": "upsert 更新后的标题",
    "category": "技术",
    "read_count": 2000
})

# 批量 upsert
data = [
    {
        "id": i,
        "article_vector": [random.random() for _ in range(128)],
        "title": f"upsert 文章_{i}",
        "category": "技术",
        "read_count": i * 10
    }
    for i in range(100)
]
collection.upsert(data)
print(f"批量 upsert {len(data)} 条数据完成")

MilvusClient 方式 ​

python
from pymilvus import MilvusClient

client = MilvusClient(uri="http://localhost:19530")

client.upsert(
    collection_name="article_search",
    data=[{"id": i, "vector": [0.1] * 128} for i in range(10)]
)

upsert 与 delete + insert 对比 ​

对比项upsertdelete + insert
主键冲突自动处理需手动判断
原子性单次操作原子完成两步操作，存在中间状态
性能更高效，一次写日志两次写入，开销更高
适用场景数据同步、幂等写入需要清理旧数据的复杂场景

数据统计 ​

获取集合统计信息 ​

python
# 获取实体数量
count = collection.num_entities
print(f"集合中的实体数量: {count}")

# 获取集合统计信息
stats = collection.get_stats()
print(f"集合统计: {stats}")

查询统计 ​

python
# 统计各分类文章数量
results = collection.query(
    expr='category == "技术"',
    output_fields=["id"]
)
tech_count = len(results)
print(f"技术类文章数量: {tech_count}")

批量导入（BulkImport） ​

当数据规模达到百万级以上时，逐条或小批量 insert 的效率会显著下降。Milvus 提供的 BulkImport（批量导入）能力，通过对象存储直接读取整文件入库，吞吐量远高于 insert，适合初始化大批量历史数据。

适用场景对比 ​

对比项insertBulkImport
数据量级千 ~ 十万级百万 ~ 亿级
数据来源内存中的 Python 对象对象存储（MinIO/S3）上的文件
速度受网络和 WAL 限制接近存储带宽，吞吐高
实时性写入即可见（取决于一致性）异步任务，完成后可见
典型用途在线写入、增量更新集群初始化、历史数据迁移

准备数据文件 ​

将数据组织为 JSON 或 Parquet 文件，并上传到 Milvus 所连接的对象存储（MinIO/S3）。文件需符合集合的 schema，例如：

json
// rows.json
[
  {
    "id": 1,
    "article_vector": [0.1, 0.2],
    "title": "文章1",
    "category": "技术"
  },
  {
    "id": 2,
    "article_vector": [0.3, 0.4],
    "title": "文章2",
    "category": "生活"
  }
]

使用 utility.do_bulk_insert ​

python
from pymilvus import connections, utility

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

# 发起批量导入任务
job_id = utility.do_bulk_insert(
    collection_name="article_search",
    files=["rows.json"]     # 对象存储上的文件路径列表，支持 JSON/Parquet
)

print(f"批量导入任务已提交，job_id={job_id}")

使用 MilvusClient 方式 ​

python
from pymilvus import MilvusClient

client = MilvusClient(uri="http://localhost:19530")

job_id = client.create_bulk_insert_job(
    collection_name="article_search",
    files=["rows.json"]
)

查询任务状态 ​

BulkImport 是异步任务，需要轮询状态直至完成：

python
from pymilvus import utility

# 轮询任务状态
while True:
    state = utility.get_bulk_insert_state(job_id)
    print(f"任务状态: {state.state}, 进度: {state.progress}")

    if state.state in ("Completed", "Failed"):
        break

    import time
    time.sleep(5)

# 列出集合的所有导入任务
jobs = utility.list_bulk_insert_tasks(collection_name="article_search")
for j in jobs:
    print(j.job_id, j.state)

完整流程 ​

python
from pymilvus import connections, utility
import time

def bulk_import_demo():
    connections.connect(host="localhost", port="19530")

    # 1. 文件已上传到对象存储（此处略去上传步骤）

    # 2. 提交导入任务
    job_id = utility.do_bulk_insert(
        collection_name="article_search",
        files=["rows.json"]
    )
    print(f"提交任务: {job_id}")

    # 3. 轮询直到完成
    while True:
        state = utility.get_bulk_insert_state(job_id)
        print(f"状态={state.state}, 进度={state.progress}")
        if state.state in ("Completed", "Failed"):
            break
        time.sleep(5)

    print("批量导入完成")

bulk_import_demo()

数据导入导出 ​

导出数据 ​

python
import json

# 查询所有数据
results = collection.query(
    expr="",
    output_fields=["id", "title", "category", "read_count"],
    limit=10000
)

# 保存为 JSON
with open("articles.json", "w", encoding="utf-8") as f:
    json.dump(results, f, ensure_ascii=False, indent=2)

print(f"已导出 {len(results)} 条数据")

导入数据 ​

python
import json

# 从 JSON 文件加载
with open("articles.json", "r", encoding="utf-8") as f:
    data = json.load(f)

# 批量插入
batch_size = 1000
for i in range(0, len(data), batch_size):
    batch = data[i:i + batch_size]
    collection.insert(batch)
    print(f"已导入第 {i//batch_size + 1} 批数据")

print(f"总共导入 {len(data)} 条数据")

批量操作最佳实践 ​

高效批量插入 ​

python
def batch_insert(collection, data_list, batch_size=1000):
    """高效批量插入数据"""
    total = len(data_list)
    for i in range(0, total, batch_size):
        batch = data_list[i:i + batch_size]
        collection.insert(batch)
        print(f"已插入 {min(i + batch_size, total)}/{total} 条数据")

# 使用示例
data_list = [...]  # 大量数据
batch_insert(collection, data_list, batch_size=1000)

批量删除 ​

python
def batch_delete(collection, ids, batch_size=100):
    """批量删除数据"""
    for i in range(0, len(ids), batch_size):
        batch_ids = ids[i:i + batch_size]
        id_str = ",".join(map(str, batch_ids))
        collection.delete(f"id in [{id_str}]")
        print(f"已删除第 {i//batch_size + 1} 批数据")

# 使用示例
ids_to_delete = [1, 2, 3, ..., 1000]
batch_delete(collection, ids_to_delete)

完整示例 ​

python
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
import random
import time

def demo_data_management():
    """数据管理完整示例"""

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

    # 创建集合
    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=256),
        FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=64),
        FieldSchema(name="score", dtype=DataType.FLOAT)
    ]

    schema = CollectionSchema(fields, "示例集合")
    collection = Collection("demo_collection", schema)

    print("=== 插入数据 ===")
    # 插入示例数据
    data = []
    categories = ["科技", "生活", "娱乐", "教育"]
    for i in range(100):
        data.append({
            "vector": [random.random() for _ in range(128)],
            "title": f"文章_{i}",
            "category": random.choice(categories),
            "score": random.uniform(0, 100)
        })

    insert_result = collection.insert(data)
    print(f"插入 {len(insert_result.primary_keys)} 条数据")

    print("\n=== 查询数据 ===")
    # 查询科技类文章
    results = collection.query(
        expr='category == "科技"',
        output_fields=["id", "title", "score"],
        limit=5
    )
    for r in results:
        print(f"ID: {r['id']}, 标题: {r['title']}, 评分: {r['score']:.2f}")

    print("\n=== 统计数据 ===")
    print(f"总数量: {collection.num_entities}")

    print("\n=== 删除数据 ===")
    # 删除评分低于 20 的数据
    collection.delete("score < 20")
    print(f"删除后数量: {collection.num_entities}")

    # 清理
    from pymilvus import utility
    utility.drop_collection("demo_collection")
    print("\n演示完成！")

if __name__ == "__main__":
    demo_data_management()

注意事项 ​

  1. 批量大小: 建议每次插入 1000-10000 条数据，过大可能导致内存问题
  2. 主键冲突: 使用 auto_id=True 避免主键冲突
  3. 数据类型: 确保插入数据的类型与 Schema 定义一致
  4. 向量维度: 向量维度必须与 Schema 中定义的 dim 一致
  5. 字符编码: 字符串字段使用 UTF-8 编码

下一步 ​

掌握数据插入与管理后，你可以：

  1. 学习向量搜索
  2. 了解索引优化
  3. 探索高级功能

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1