Redis 集群数据迁移：不停服的前提下把向量索引迁移到新集群的方案

Redis 集群数据迁移不停服的前提下把向量索引迁移到新集群的方案一、深度引言与场景痛点大家好我是赵咕咕。有一段时间我对停服迁移这件事产生了 PTSD。每次产品经理小心翼翼地问这个数据迁移要停多久我就知道接下来要么是凌晨加班要么是被用户骂。最近一次是迁移向量索引——从旧 Redis Cluster 迁到新集群。向量索引跟普通 K-V 数据最大的区别是向量数据量大每条向量 768 维 × 4 字节 3KB百万条就是 3TB而且索引构建耗时FLAT 索引还快HNSW 索引重建可能要几十分钟。传统做法是停服、导出 RDB、导入新集群、重建索引。这一套下来至少 2 小时——对 7×24 的 RAG 服务来说完全不可接受。这篇文章分享一套不停服迁移的方案核心思路是双写 渐进式数据同步 流量切换。在迁移过程中线上服务照常运行用户无感知。二、底层机制与原理深度剖析2.1 不停服迁移的核心挑战迁移向量索引比迁移普通 Redis 数据难在哪数据量大普通 KV 可能一条几百字节向量数据一条几 KB百万级就是 GB 级别。全量同步的带宽和时间都很可观。索引重建慢Redis Stack 的向量索引FLAT/HNSW在大数据量下重建可能花费几十分钟。如果迁移过程中服务还要写入索引必须在持续写入的情况下保持可用。一致性要求迁移期间同一个 key 可能在旧集群和新集群中同时存在但值不同用户在迁移过程中更新了数据。必须保证最终一致。切换风险流量从旧集群切到新集群是一个惊险一跳。如果新集群有问题必须能秒级回滚。2.2 双写 渐进式同步的架构设计整个迁移分为四个阶段Phase 1 - 双写开启应用层同时往旧集群和新集群写入。这一步不改变读路径只是让新集群开始积累数据。双写异常不阻塞主流程——新集群暂时写失败没关系后续同步会补齐。Phase 2 - 全量同步同步服务从旧集群遍历所有 keySCAN用DUMP/RESTORE批量迁移存量数据。向量索引暂时不在新集群重建——数据先以纯 KV 形式迁过去。Phase 3 - 增量追平全量同步结束后新集群的数据还差双写开启前的几秒钟。通过对比两个集群的数据量或抽样校验追平最后的差异。Phase 4 - 流量切换读流量按比例逐步切到新集群10% → 50% → 100%。监控延迟、错误率和向量检索准确率。一旦异常秒级回滚到旧集群。2.3 向量索引的渐进式构建迁移中最耗时的环节是在新集群上重建向量索引。HNSW 索引在百万级数据下的构建需要遍历所有向量计算图结构不能跳过。优化方案是分批次渐进构建全量同步完成后不立即在新集群上运行FT.CREATE。先在新集群上以NOINDEX模式写入所有向量数据。然后分 10 批每批 10 万条逐步用FT.CREATE构建索引。每批之间暂停 30 秒让索引构建的 CPU 开销自然消化。这样新集群的 CPU 不会突然打满服务也始终可用。三、生产级代码实现import asyncio import logging from dataclasses import dataclass, field from typing import AsyncIterator import aioredis logger logging.getLogger(__name__) dataclass class MigrationProgress: 迁移进度追踪。 total_keys: int 0 synced_keys: int 0 failed_keys: int 0 phase: str idle errors: list[str] field(default_factorylist) class RedisVectorMigration: Redis 向量索引不停服迁移器。 def __init__( self, old_redis_url: str, new_redis_url: str, batch_size: int 500, parallel_workers: int 4, ): self._old_url old_redis_url self._new_url new_redis_url self._batch batch_size self._workers parallel_workers self._old: aioredis.Redis | None None self._new: aioredis.Redis | None None self._progress MigrationProgress() async def connect(self) - None: 建立两个集群的连接。 self._old aioredis.from_url( self._old_url, max_connections20, socket_timeout10, retry_on_timeoutTrue, ) self._new aioredis.from_url( self._new_url, max_connections20, socket_timeout10, retry_on_timeoutTrue, ) # 检查连接 await asyncio.gather(self._old.ping(), self._new.ping()) logger.info(两个 Redis 集群连接成功) async def run_full_migration(self) - MigrationProgress: 执行完整的不停服迁移流程。 # Phase 1: 双写由应用层负责此处跳过 # Phase 2: 全量同步 self._progress.phase full_sync await self._full_sync() # Phase 3: 增量追平 self._progress.phase incremental_sync for _ in range(6): # 最多追 6 轮 diff await self._incremental_sync() if diff 0: logger.info(增量已追平差异为 0) break await asyncio.sleep(10) # Phase 4: 数据校验 self._progress.phase validation consistent await self._validate_consistency() if not consistent: logger.error(数据校验未通过请人工介入) self._progress.errors.append(一致性校验失败) self._progress.phase completed return self._progress async def _full_sync(self) - None: 全量扫描旧集群并同步到新集群。 cursor 0 while True: cursor, keys await self._old.scan( cursor, matchvec:*, countself._batch ) if keys: await self._batch_restore(keys) self._progress.synced_keys len(keys) logger.info( 全量同步进度: %d keys, cursor%d, self._progress.synced_keys, cursor, ) if cursor 0: break await asyncio.sleep(0.01) # 让出事件循环 self._progress.total_keys self._progress.synced_keys logger.info(全量同步完成: %d keys, self._progress.total_keys) async def _batch_restore(self, keys: list[bytes]) - None: 批量 DUMP RESTORE。 # 分批处理每批 parallel_workers 个并发 batches [ keys[i:i self._workers] for i in range(0, len(keys), self._workers) ] for batch in batches: tasks [self._migrate_one_key(k) for k in batch] results await asyncio.gather(*tasks, return_exceptionsTrue) for key, result in zip(batch, results): if isinstance(result, Exception): logger.error(迁移 key %s 失败: %s, key, result) self._progress.failed_keys 1 self._progress.errors.append(str(result)) async def _migrate_one_key(self, key: bytes) - None: 迁移单个 keyDUMP PTTL RESTORE。 pipe self._old.pipeline() pipe.dump(key) pipe.pttl(key) serialized_val, ttl_ms await pipe.execute() if serialized_val is None: logger.warning(key %s 在旧集群中不存在跳过, key) return try: if ttl_ms and ttl_ms 0: await self._new.restore(key, ttl_ms, serialized_val, replaceTrue) else: await self._new.restore(key, 0, serialized_val, replaceTrue) except aioredis.ResponseError as e: if BUSYKEY in str(e): # key 已存在双写产生的用 replaceTrue 覆盖 await self._new.restore(key, 0, serialized_val, replaceTrue) else: raise async def _incremental_sync(self) - int: 增量同步对比两个集群的数据量差异并补齐。 old_size await self._old.dbsize() new_size await self._new.dbsize() diff old_size - new_size if diff 0: return 0 logger.info(增量差异: %d keys, diff) # 抽样校验差异的 key cursor 0 checked 0 missing: list[bytes] [] while checked min(diff * 5, 10000): cursor, keys await self._old.scan( cursor, matchvec:*, count100 ) if not keys: break pipe self._new.pipeline() for k in keys: pipe.exists(k) exists_results await pipe.execute() for k, exists in zip(keys, exists_results): if not exists: missing.append(k) checked 1 if cursor 0: break if missing: await self._batch_restore(missing) self._progress.synced_keys len(missing) return diff async def _validate_consistency(self) - bool: 随机抽样校验数据一致性。 import random # 在新集群随机抽 1000 个 key cursor 0 sample_keys: list[bytes] [] while len(sample_keys) 1000: cursor, keys await self._new.scan( cursor, matchvec:*, count100 ) sample_keys.extend(keys) if cursor 0: break sample_keys random.sample( sample_keys, min(1000, len(sample_keys)) ) mismatched 0 for key in sample_keys: try: val_old await self._old.dump(key) val_new await self._new.dump(key) if val_old ! val_new: mismatched 1 logger.warning(数据不一致: %s, key) except Exception as e: logger.error(校验 key %s 失败: %s, key, e) mismatched 1 match_rate 1 - mismatched / max(len(sample_keys), 1) logger.info(数据一致性: %.2f%% (%d/%d mismatched), match_rate * 100, mismatched, len(sample_keys)) return match_rate 0.999 async def close(self) - None: 关闭连接。 for conn in [self._old, self._new]: if conn: await conn.close() # ─── 双写装饰器应用层使用 ─── class DualWriteWrapper: 在应用层实现双写。 def __init__(self, primary: aioredis.Redis, secondary: aioredis.Redis): self._primary primary self._secondary secondary async def set_vec(self, key: str, vector: bytes, ttl: int 0) - bool: 双写到两个集群。 # 主集群写入阻塞必须成功 result await self._primary.set(key, vector, exttl) # 副集群写入异步失败不阻塞 asyncio.create_task(self._safe_write_secondary(key, vector, ttl)) return result async def _safe_write_secondary( self, key: str, vector: bytes, ttl: int ) - None: try: await asyncio.wait_for( self._secondary.set(key, vector, exttl), timeout2.0, ) except Exception as e: logger.warning(双写副集群失败 (key%s): %s, key[:20], e) async def main(): migrator RedisVectorMigration( old_redis_urlredis://old-cluster:6379, new_redis_urlredis://new-cluster:6379, batch_size500, parallel_workers4, ) await migrator.connect() try: progress await migrator.run_full_migration() print(f迁移完成: {progress.synced_keys} keys 已同步) if progress.errors: print(f错误: {len(progress.errors)} 条) finally: await migrator.close() if __name__ __main__: asyncio.run(main())实现要点SCAN 代替 KEYSRedis 的KEYS命令会阻塞整个实例。SCAN是游标式迭代每次只返回少量 key不会阻塞其他请求。DUMP/RESTORE 序列化DUMP导出的二进制格式包含完整的 key 数据含类型和过期时间RESTORE直接还原。比GETSET更准确——尤其对于带 TTL 的 key。Pipeline 批量操作迁移时用 Pipeline 批量发送命令减少网络 RTT。_migrate_one_key里dump和pttl合并到一个 pipeline 里执行。双写异常不阻塞_safe_write_secondary用create_task异步执行且兜底异常副集群写失败不会影响用户请求。靠后续增量同步补齐差异。渐进式校验校验阶段随机抽样 1000 个 key 做DUMP对比。99.9% 一致率达标即可。四、边界分析与架构权衡4.1 双写的一致性问题双写模式在最坯情况下可能出现主集群成功、副集群失败的情况。这是最终一致性方案不保证强一致。如果业务要求严格一致性需要在应用层实现两阶段提交或分布式事务代价极高。对于向量检索场景最终一致性通常可接受一条向量晚几秒出现在新集群不影响检索结果只要旧集群还能查到就行。4.2 向量索引重建的时机这是一个关键的时序决策选项 A先迁数据切换流量后再建索引。优点是索引只建一次缺点是切换后服务需要等待索引构建完成几十分钟不可用。选项 B迁移过程中就在新集群建索引。优点是切换后立即可用缺点是需要额外的 CPU/内存来维护索引。推荐选项 B 的变体迁移 80% 数据后开始在新集群构建索引利用剩余的 20% 数据做索引的增量更新。这样既能缩短切换后的等待时间又能保持索引接近完整。4.3 Redis Stack 向量索引的特殊处理Redis Stack 的向量索引通过FT.CREATE创建索引元数据存储在 RediSearch 模块中不受DUMP/RESTORE影响。这意味着你需要在目标集群单独创建索引定义。建议在迁移脚本中增加一个_create_index_if_not_exists方法在迁移开始前检查新集群的索引是否存在不存在则创建async def _ensure_index(self, index_name: str, dim: int, metric: str COSINE): try: await self._new.ft(index_name).info() except aioredis.ResponseError: await self._new.ft(index_name).create_index( schema( aioredis.commands.search.VectorField( embedding, HNSW, { TYPE: FLOAT32, DIM: dim, DISTANCE_METRIC: metric, M: 16, EF_CONSTRUCTION: 200, } ), ) ) logger.info(新集群向量索引已创建: %s (dim%d), index_name, dim)4.4 回滚方案流量切换阶段必须保留旧集群的运行。回滚只需一步把读流量的比例切回 0%。不需要数据回滚——因为旧集群的数据在切换期间一直是最全的。建议把流量比例放在配置中心如 Nacos/Consul这样回滚不需要重启服务秒级生效。五、总结不停服迁移的本质是用时间换可用性——用几小时甚至几天的渐进式同步换来 0 秒的服务中断。这套方案的核心三板斧双写 异步兜底新集群写入失败不阻塞主流程靠增量同步补齐差异。SCAN DUMP/RESTORE全量同步走 Redis 原生序列化高效且类型安全。渐进切换 秒级回滚读流量按比例切配置中心控制出问题秒级切回。迁移完记得删掉双写逻辑。双写是手段不是目的长期双写是问题不是方案。下一篇预告Prompt 敏感信息脱敏在发给 LLM 之前自动识别并替换 PII 和密钥。