4.1 测试环境搭建与数据集准备

测试环境规划

为了确保测试结果的可比性和准确性，我们需要构建标准化的测试环境。

硬件配置

基准配置：

• CPU: Intel Xeon Silver 4210R (10核20线程)
• 内存: 64GB DDR4 ECC
• 存储: 2TB NVMe SSD
• 网络: 10Gbps以太网
• GPU: NVIDIA Tesla T4 (可选)

扩展配置（用于大规模测试）：

• CPU: Intel Xeon Gold 6248R (24核48线程)
• 内存: 128GB DDR4 ECC
• 存储: 4TB NVMe SSD RAID 0
• 网络: 25Gbps以太网
• GPU: NVIDIA A100 40GB

软件环境

操作系统：

# Ubuntu 20.04 LTS
sudo apt update
sudo apt upgrade -y
sudo apt install -y build-essential cmake git python3-pip
​

依赖软件：

# Python依赖
pip3 install numpy pandas matplotlib seaborn scikit-learn
pip3 install milvus-client qdrant-client weaviate-client
pip3 install pytest pytest-benchmark memory-profiler

# 系统优化
sudo sysctl -w vm.swappiness=1
echo "* soft nofile 65536" | sudo tee -a /etc/security/limits.conf
echo "* hard nofile 65536" | sudo tee -a /etc/security/limits.conf
​

数据集准备

测试数据集的选择对测试结果有重要影响。我们使用多种标准数据集进行测试。

文本数据集

SST-2 (Stanford Sentiment Treebank)：

• 训练集：67,349句
• 测试集：8,742句
• 维度：768维（BERT-base）
• 文件大小：约100MB

代码数据集：

import numpy as np
from datasets import load_dataset

# 加载SST-2数据集
dataset = load_dataset("sst2")
train_texts = dataset['train']['sentence']
train_labels = dataset['train']['label']

# 生成嵌入向量
embeddings = []
for text in train_texts:
    embedding = generate_embedding(text)  # 使用BERT生成嵌入
    embeddings.append(embedding)
embeddings = np.array(embeddings)

# 保存数据集
np.save('sst2_embeddings.npy', embeddings)
np.save('sst2_labels.npy', train_labels)
​

图像数据集

CIFAR-10：

• 图像数量：60,000张
• 图像尺寸：32×32×3
• 维度：2048维（ResNet-18）
• 文件大小：约500MB

图像数据处理：

from PIL import Image
import torch
import torchvision.transforms as transforms
from torchvision.models import resnet18

# 加载预训练模型
model = resnet18(pretrained=True)
model.eval()

# 图像预处理
transform = transforms.Compose([
    transforms.Resize((32, 32)),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])

def extract_image_embedding(image_path):
    image = Image.open(image_path)
    image_tensor = transform(image).unsqueeze(0)
    
    with torch.no_grad():
        embedding = model(image_tensor)
    
    return embedding.squeeze().numpy()
​

混合数据集

为了模拟真实场景，我们创建混合数据集：

• 70% 文本数据：SST-2
• 20% 图像数据：CIFAR-10
• 10% 代码数据：GitHub代码片段

# 创建混合数据集
def create_hybrid_dataset():
    # 加载各类数据
    text_embeddings = np.load('sst2_embeddings.npy')
    image_embeddings = np.load('cifar10_embeddings.npy')
    code_embeddings = np.load('code_embeddings.npy')
    
    # 合并数据
    all_embeddings = np.vstack([
        text_embeddings,
        image_embeddings,
        code_embeddings
    ])
    
    # 保存完整数据集
    np.save('hybrid_embeddings.npy', all_embeddings)
    
    # 生成查询向量
    query_vectors = generate_query_vectors(all_embeddings, 1000)
    np.save('query_vectors.npy', query_vectors)
    
    return all_embeddings, query_vectors

def generate_query_vectors(embeddings, num_queries):
    """生成查询向量"""
    np.random.seed(42)
    indices = np.random.choice(len(embeddings), num_queries, replace=True)
    return embeddings[indices]
​

测试工具链

我们使用完整的测试工具链来执行性能测试。

测试框架

# benchmark_suite.py
import time
import psutil
import numpy as np
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class TestResult:
    database: str
    operation: str
    latency_ms: float
    qps: float
    memory_usage_mb: float
    cpu_usage_percent: float
    error_rate: float

class VectorDatabaseBenchmark:
    def __init__(self, database_configs: Dict[str, Any]):
        self.databases = database_configs
        self.results = []
        
    def setup_database(self, db_name: str):
        """初始化数据库连接"""
        config = self.databases[db_name]
        if db_name == "milvus":
            self.client = MilvusClient(**config)
        elif db_name == "qdrant":
            self.client = QdrantClient(**config)
        elif db_name == "weaviate":
            self.client = WeaviateClient(**config)
    
    def load_test_data(self, embeddings: np.ndarray, queries: np.ndarray):
        """加载测试数据"""
        collection_name = "benchmark_test"
        
        # 创建集合
        self.create_collection(collection_name, embeddings.shape[1])
        
        # 分批插入数据
        batch_size = 1000
        for i in range(0, len(embeddings), batch_size):
            batch = embeddings[i:i + batch_size]
            self.insert_batch(collection_name, batch)
    
    def run_latency_test(self, queries: np.ndarray, num_trials: int = 1000):
        """运行延迟测试"""
        latencies = []
        
        for i, query in enumerate(queries[:num_trials]):
            start_time = time.time()
            results = self.search(query, top_k=10)
            end_time = time.time()
            
            latency = (end_time - start_time) * 1000  # 转换为毫秒
            latencies.append(latency)
            
            # 记录系统资源使用
            memory = psutil.Process().memory_info().rss / 1024 / 1024
            cpu = psutil.Process().cpu_percent()
        
        return np.array(latencies)
    
    def run_throughput_test(self, queries: np.ndarray, duration_seconds: int = 60):
        """运行吞吐量测试"""
        start_time = time.time()
        end_time = start_time + duration_seconds
        
        query_count = 0
        successful_queries = 0
        
        while time.time() < end_time:
            query = queries[np.random.randint(len(queries))]
            try:
                self.search(query, top_k=10)
                successful_queries += 1
            except Exception as e:
                pass
            query_count += 1
        
        actual_duration = time.time() - start_time
        qps = successful_queries / actual_duration
        
        return qps
​