docker部署milvus
·

1.拉取镜像
docker pull swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/milvusdb/milvus:v2.4.5
docker pull swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/minio/minio:RELEASE.2023-03-20T20-16-18Z
2.创建docker-compose.yml文件(支持 Milvus v2.4.5 单机模式)
version: '3.5'
services:
etcd:
container_name: milvus-etcd
image: quay.io/coreos/etcd:v3.5.5
environment:
- ETCD_AUTO_COMPACTION_MODE=revision
- ETCD_AUTO_COMPACTION_RETENTION=1000
- ETCD_QUOTA_BACKEND_BYTES=4294967296
- ETCD_SNAPSHOT_COUNT=50000
- ETCD_HEARTBEAT_INTERVAL=500
- ETCD_ELECTION_TIMEOUT=2500
- ETCD_LISTEN_CLIENT_URLS=http://0.0.0.0:2379
- ETCD_ADVERTISE_CLIENT_URLS=http://etcd:2379
volumes:
- etcd_data:/etcd
networks:
- milvus-network
minio:
container_name: milvus-minio
image: swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/minio/minio:RELEASE.2023-03-20T20-16-18Z
environment:
- MINIO_ACCESS_KEY=minioadmin
- MINIO_SECRET_KEY=minioadmin
command: server /minio_data
volumes:
- minio_data:/minio_data
ports:
- "9000:9000"
networks:
- milvus-network
standalone:
container_name: milvus-standalone
image: swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/milvusdb/milvus:v2.4.5
# command: ["milvus", "run", "standalone"]
command: ["/bin/sh", "-c", "sleep 5 && milvus run standalone"]
depends_on:
- etcd
- minio
ports:
- "19530:19530" # Milvus gRPC 接口
- "9091:9091" # Milvus HTTP 接口(用于监控等)
environment:
- ETCD_ENDPOINTS=etcd:2379
- MINIO_ADDRESS=minio:9000
- MINIO_ACCESS_KEY=minioadmin
- MINIO_SECRET_KEY=minioadmin
- MILVUS_LOG_LEVEL=debug
volumes:
- milvus_data:/var/lib/milvus
networks:
- milvus-network
volumes:
etcd_data:
minio_data:
milvus_data:
networks:
milvus-network:
3.启动 Milvus
docker-compose up -d
首次执行会自动拉取所有相关镜像(大概几百 MB 到几个 GB),视网速而定。完成后,你会看到类似:
Creating milvus-etcd ... done
Creating milvus-minio ... done
Creating milvus-standalone ... done
并且你可以通过docker ps命令查看容器运行情况
docker ps --format "table {{.ID}}\t{{.Names}}\t{{.Image}}\t{{.Status}}"
4.验证三组件协同工作的建议测试方案
我们用 PyMilvus 做个完整流程测试(创建集合 → 插入数据 → 搜索 → 删除集合),只要这些操作能顺利完成,就可以确认这三种服务已经协同运行。
🧪 步骤一:准备 Python 环境
确保你已经安装了 PyMilvus:
pip install pymilvus -i https://mirrors.aliyun.com/pypi/simple
🧪 步骤二:连接到 Milvus
打开 Python 控制台或写一个脚本:
from pymilvus import connections
# 默认连接 milvus-standalone 的 19530 端口
connections.connect("default", host="localhost", port="19530")
如果这里没有报错,说明 milvus-standalone 正常运行。
🧪 步骤三:创建向量集合(Collection)
from pymilvus import FieldSchema, CollectionSchema, DataType, Collection
# 定义字段(向量 + 主键)
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=4)
]
schema = CollectionSchema(fields, description="Test collection")
collection = Collection(name="demo_collection", schema=schema)
print("✅ 集合创建成功")
🧪 步骤四:插入向量数据
import random
data = [
[i for i in range(10)], # 主键 ID
[[random.random() for _ in range(4)] for _ in range(10)] # 向量
]
collection.insert(data)
collection.flush() # 强制写入 MinIO
print("✅ 数据插入成功")
🧪 步骤五:创建索引
index_params = {
"metric_type": "L2",
"index_type": "IVF_FLAT",
"params": {"nlist": 64}
}
collection.create_index(field_name="embedding", index_params=index_params)
print("✅ 索引创建成功")
🧪 步骤六:搜索向量
collection.load()
results = collection.search(
data=[[0.1, 0.2, 0.3, 0.4]],
anns_field="embedding",
param={"nprobe": 10},
limit=3,
output_fields=["id"]
)
for hits in results:
for hit in hits:
print(f"相似向量ID: {hit.id}, 距离: {hit.distance}")
如果你看到正常返回的结果,说明 Milvus 成功完成:
- 使用 etcd 协调元数据
- 使用 minio 存储数据和索引
- 使用 standalone 完成计算和检索
🧪 步骤七(可选):删除集合以测试清理
collection.drop()
print("✅ 集合已删除")
完整测试脚本
from pymilvus import (
connections, FieldSchema, CollectionSchema, DataType, Collection, utility
)
import random
# ---------------------------
# Step 1: 连接 Milvus 服务
# ---------------------------
connections.connect("default", host="localhost", port="19530")
print("[1] 成功连接 Milvus")
# ---------------------------
# Step 2: 定义集合名称与维度
# ---------------------------
collection_name = "test_vector_collection"
dim = 4 # 向量维度
# ---------------------------
# Step 3: 如果集合存在则删除,避免重复创建报错
# ---------------------------
if utility.has_collection(collection_name):
collection = Collection(name=collection_name)
collection.drop()
print(f"[2] 已存在集合 {collection_name},已删除")
# ---------------------------
# Step 4: 定义字段结构并创建集合
# ---------------------------
fields = [
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim)
]
schema = CollectionSchema(fields, description="用于测试的向量集合")
collection = Collection(name=collection_name, schema=schema)
print(f"[3] 成功创建集合 {collection_name}")
# ---------------------------
# Step 5: 构造向量并插入数据
# ---------------------------
vectors = [[random.random() for _ in range(dim)] for _ in range(5)]
ids = [i for i in range(5)]
data = [ids, vectors]
collection.insert(data)
print("[4] 成功插入数据")
print("插入的向量:", vectors)
# ---------------------------
# Step 6: 创建索引(必须)
# ---------------------------
index_params = {
"metric_type": "L2", # 距离度量方式
"index_type": "IVF_FLAT", # 索引类型
"params": {"nlist": 64} # IVF 参数
}
collection.create_index(field_name="embedding", index_params=index_params)
print("[5] 成功创建索引")
# ---------------------------
# Step 7: 加载集合到内存
# ---------------------------
collection.load()
print("[6] 成功加载集合")
# ---------------------------
# Step 8: 搜索测试(使用第一条插入向量)
# ---------------------------
results = collection.search(
data=[vectors[0]], # 查询向量
anns_field="embedding", # 用于 ANN 搜索的字段
param={"nprobe": 10}, # 搜索参数(用于 IVF)
limit=3, # 返回 Top-3 结果
output_fields=["id"] # 返回的附加字段
)
print("[7] 成功搜索,结果如下:")
for result in results[0]:
print(f"ID: {result.entity.get('id')}, distance: {result.distance}")
命令行运行测试代码
python3 milvus_test.py
示例输出
[1] 成功连接 Milvus
[2] 已存在集合 test_vector_collection,已删除
[3] 成功创建集合 test_vector_collection
[4] 成功插入数据
插入的向量: [[...], [...], ...]
[5] 成功创建索引
[6] 成功加载集合
[7] 成功搜索,结果如下:
ID: 0, distance: 0.0
ID: 3, distance: 0.8123
ID: 1, distance: 0.9285
更多推荐



所有评论(0)