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

Docker Compose 运行 Milvus (Mac) 并与 python 连接测试

news 2025/9/10 10:12:55

Milvus 简单来说是一个向量化数据库，有 lite、standalone、distributed 三个版本，其中 lite 轻量化版本直接安装 python SDK 运行即可（pip install pymilvus），本文讲述如何借助 docker，在 mac 环境下配置并建立与 python 的连接。

安装 Docker

在官网下载 dmg 包 Mac | Docker Docs

看清楚自己的 mac 是什么芯片，m系列就和我一样选第一个

下载后跟其他软件一样安装即可，安装完成后打开终端，输入 docker -v 验证是否安装成功

下载 Docker 软件会看到类似的界面（我这里是运行了一些容器，大家刚安装正常是没有东西的）

安装 Milvus

拉取 docker compose 文件

wget https://github.com/milvus-io/milvus/releases/download/v2.5.14/milvus-standalone-docker-compose.yml -O docker-compose.yml

在终端输入上面的指令并回车，会下载一个 ocker-compose.yml

启动 Milvus

在终端输入下面的指令

docker compose up -d

运行后打开刚才下载的 docker 软件，就开始看到名为milvus- standalone、milvus-minio 和milvus-etcd的容器已经启动。

简单介绍一下这三个组件（截取自官网）

milvus-etcd容器不向主机暴露任何端口，并将其数据映射到当前文件夹中的volumes/etcd。
milvus-minio容器使用默认身份验证凭据在本地为端口9090和9091提供服务，并将其数据映射到当前文件夹中的volumes/minio。
Milvus-standalone容器使用默认设置为本地19530端口提供服务，并将其数据映射到当前文件夹中的volumes/milvus。

在终端运行 docker-compose ps 也能看到当前容器状态

Python 连接测试

首先在终端运行下面指令安装 python sdk（sdk 其实就是个工具包）

pip install pymilvus==2.6.0b0

如果 PyMilvus 安装正确，运行以下命令时不会出现异常。

python3 -c "from pymilvus import Collection"

贴出我的测试代码

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Milvus 连接测试代码
演示基本的连接、创建集合、插入数据、搜索等功能
"""import time
import random
import numpy as np
from pymilvus import (
connections,
utility,
FieldSchema,
CollectionSchema,
DataType,
Collection,
)def test_milvus_connection():"""测试Milvus连接"""print("正在连接Milvus...")try:# 连接到Milvus服务器connections.connect(alias="default",  # 连接别名host="localhost",  # 主机地址port="19530"       # 端口号)print("Milvus连接成功！")return Trueexcept Exception as e:print(f"Milvus连接失败: {e}")return Falsedef create_collection():"""创建测试集合"""print("\n正在创建测试集合...")# 定义字段模式fields = [FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=128),  # 128维向量FieldSchema(name="label", dtype=DataType.VARCHAR, max_length=100)]# 创建集合模式schema = CollectionSchema(fields=fields,description="测试集合 - 用于演示Milvus功能")# 创建集合collection_name = "test_collection"collection = Collection(name=collection_name,schema=schema,using="default")# 创建索引print("正在创建向量索引...")index_params = {"metric_type": "L2",           # 距离度量类型：L2欧几里得距离"index_type": "IVF_FLAT",      # 索引类型：倒排文件"params": {"nlist": 128}       # 聚类中心数量}collection.create_index(field_name="vector",index_params=index_params)print(f"集合 '{collection_name}' 创建成功！")return collectiondef insert_sample_data(collection):"""插入示例数据"""print("\n正在插入示例数据...")# 生成随机向量数据num_entities = 1000vectors = np.random.rand(num_entities, 128).astype(np.float32)labels = [f"样本_{i}" for i in range(num_entities)]# 准备插入数据entities = [vectors,  # 向量数据labels    # 标签数据]# 插入数据collection.insert(entities)collection.flush()  # 确保数据持久化print(f"成功插入 {num_entities} 条数据！")return vectorsdef search_similar_vectors(collection, query_vector):"""搜索相似向量"""print("\n正在搜索相似向量...")# 加载集合到内存collection.load()# 搜索参数search_params = {"metric_type": "L2","params": {"nprobe": 10}}# 执行搜索results = collection.search(data=[query_vector],anns_field="vector",param=search_params,limit=5,  # 返回前5个最相似的结果output_fields=["label"])print("搜索结果:")for i, hits in enumerate(results):print(f"查询向量 {i + 1}:")for j, hit in enumerate(hits):print(f"  排名 {j + 1}: ID={hit.id}, 距离={hit.distance:.4f}, 标签={hit.entity.get('label')}")return resultsdef get_collection_stats(collection):"""获取集合统计信息"""print("\n集合统计信息:")# 获取实体数量num_entities = collection.num_entitiesprint(f"  实体数量: {num_entities}")# 获取索引信息index_info = collection.index().paramsprint(f"  索引类型: {index_info.get('index_type', 'N/A')}")print(f"  距离度量: {index_info.get('metric_type', 'N/A')}")def cleanup(collection_name):"""清理测试数据"""print(f"\n正在清理测试集合 '{collection_name}'...")try:utility.drop_collection(collection_name)print(f"集合 '{collection_name}' 已删除")except Exception as e:print(f"清理时出现警告: {e}")def main():"""主函数 - 执行完整的测试流程"""print("开始Milvus功能测试演示")print("=" * 50)# 1. 测试连接if not test_milvus_connection():return# 2. 创建集合collection = create_collection()# 3. 插入示例数据vectors = insert_sample_data(collection)# 4. 获取统计信息get_collection_stats(collection)# 5. 搜索相似向量# 使用第一个向量作为查询向量query_vector = vectors[0].tolist()search_similar_vectors(collection, query_vector)# 6. 清理测试数据（可选）# cleanup("test_collection")print("\nMilvus功能测试完成！")print("提示：如需保留测试数据，请注释掉cleanup函数调用")if __name__ == "__main__":main()

运行效果：