_{🧠 记忆锚点：先快照回填，再增量追平；双写、影子比对、灰度切流，旧库保留可回滚。}

💡 答案要点

迁移六步流程（零停机）：

展开 Python 代码示例（40 行）

# Step 1: 数据盘点
stats = source_db.describe_index_stats()
# 记录 total_vectors, dimension, metadata_fields

# Step 2: 目标库初始化（复现索引参数）
qdrant_client.create_collection(
    "migrated",
    vectors_config={"size": stats["dimension"], "distance": "Cosine"}
)

# Step 3: 批量迁移（带断点续传，防止中断丢失）
def migrate_batch(source, target, batch_size=1000):
    start = load_checkpoint("ckpt.json")
    for offset in range(start, source.count(), batch_size):
        batch = source.fetch(source.list_ids(offset=offset, limit=batch_size))
        # Pinecone → Qdrant 格式转换
        points = [
            {"id": v["id"], "vector": v["values"], "payload": v.get("metadata", {})}
            for v in batch["vectors"].values()
        ]
        target.upsert(points=points)
        save_checkpoint("ckpt.json", offset + batch_size)

# Step 4: 数据验证（随机抽样 1000 条对比）
def validate(source, target, n=1000):
    for vid in source.random_sample(n):
        src_vec = source.fetch([vid]).values
        tgt_vec = target.retrieve([vid]).vector
        assert np.allclose(src_vec, tgt_vec, atol=1e-5), f"向量不匹配: {vid}"

# Step 5: 双写模式（写两库，读优先新库）
def write(vector, meta):
    source.upsert(vector, meta)   # 继续写旧库
    target.upsert(vector, meta)   # 同步写新库，防止新数据丢失

# Step 6: 灰度切换（10% → 50% → 100%）
def route(query_vector, traffic_pct=0.1):
    if random.random() < traffic_pct:
        return target.search(query_vector)  # 新库
    return source.search(query_vector)      # 旧库

常见迁移坑：

问题	解决方案
浮点精度误差	用 np.allclose(atol=1e-5) 验证
元数据类型不兼容	提前做字段类型映射表
中断导致部分迁移	断点续传（checkpoint 文件）
迁移期间新数据丢失	双写模式
索引参数未迁移	手动重建并确认 M/efConstruction

面试话术：

"向量库迁移核心是'零停机 + 数据一致性'。关键是双写模式：迁移期间同时写源库和目标库，防止新数据丢失；灰度切换（10→50→100%）逐步验证；全程断点续传防止中断丢失。我 Pinecone 迁 Qdrant 200 万条数据，双写过渡 3 天，零停机完成。"

---