attu向量数据库数据插入实战：从基础操作到性能优化

1次阅读

共计 2196 个字符，预计需要花费 6 分钟才能阅读完成。

与传统 SQL 数据库相比，向量数据库的插入操作有三大特殊性：

数据类型复杂：不仅要处理常规字段，还需校验向量维度和数据类型
性能敏感：高维向量插入容易成为性能瓶颈
资源消耗大：单条向量可能占用数 KB 内存，批量操作需特殊处理

先用最简单的单条插入熟悉基本流程。以下是 Python 示例代码：

from attu_client import VectorClient
import numpy as np

# 初始化客户端
client = VectorClient(
    host='localhost',
    port=8080,
    collection_name='demo_collection'
)

# 构造单条数据
data = {
    "id": "doc_001",
    "vector": np.random.rand(512).tolist(),  # 512 维随机向量
    "metadata": {"category": "technology"}
}

# 执行插入
try:
    res = client.insert(data)
    print(f"插入成功，ID: {res['inserted_id']}")
except Exception as e:
    print(f"插入失败: {str(e)}")

关键参数说明：

vector：必须转换为 Python 原生 list 类型
metadata：建议使用字典存储结构化数据

实际生产环境推荐使用批量插入，效率可提升 10-100 倍：

# 批量插入示例
def batch_insert(vectors, batch_size=500):
    total = len(vectors)
    for i in range(0, total, batch_size):
        batch = vectors[i:i + batch_size]
        try:
            with client.transaction():  # 启用事务
                client.batch_insert(batch)
            print(f'已插入: {min(i+batch_size, total)}/{total}')
        except Exception as e:
            print(f'批次 {i} 插入失败: {e}')
            # 这里可以添加重试逻辑

# 生成测试数据
vectors = [{"id": f"doc_{i}",
    "vector": np.random.rand(512).tolist(),
    "metadata": {"index": i}
} for i in range(10000)]

# 执行批量插入
batch_insert(vectors)

优化要点：

batch_size建议值：
512 维向量：300-800
128 维向量：1000-2000
务必使用事务保证原子性
建议添加进度打印方便监控

# 推荐配置
client = VectorClient(
    ...,
    pool_size=20,  # 根据 QPS 调整
    pool_timeout=10  # 秒
)

处理大向量时：

避免在内存中堆积所有待插入数据
使用生成器逐批产生数据
监控进程内存使用量

# 内存友好型数据生成
def vector_generator(total):
    for i in range(total):
        yield {"id": f"doc_{i}",
            "vector": np.random.rand(512).tolist()}

推荐指数退避重试算法：

import time
from random import random

def safe_insert(data, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.insert(data)
        except Exception as e:
            wait = min((2 ** attempt) + random(), 10)  # 上限 10 秒
            time.sleep(wait)
    raise Exception(f"插入失败，已达最大重试次数{max_retries}")

请使用以下规范生成测试数据：

向量维度：512
数据量：10,000 条
元数据包含：
自增 ID
时间戳（模拟真实数据）

实现批量插入并记录总耗时
尝试不同 batch_size 比较性能
故意制造维度不匹配错误并处理

# 参考答案框架
import time

def benchmark():
    # 数据生成
    dim = 512
    docs = [...]  # 10,000 条数据

    # 性能测试
    for bs in [200, 500, 1000]:
        start = time.time()
        batch_insert(docs, batch_size=bs)
        print(f'batch_size={bs}, 耗时: {time.time()-start:.2f}s')

    # 异常处理测试
    bad_doc = {"id": "bad", "vector": [1]*128}  # 错误维度
    try:
        client.insert(bad_doc)
    except Exception as e:
        print(f'成功捕获异常: {e}')