93 lines
4.8 KiB
Markdown
93 lines
4.8 KiB
Markdown
---
|
||
id: scenario.data-sync
|
||
title: 数据同步：业务库到搜索引擎 / 向量库 / 报表库，outbox 与 CDC、幂等、顺序、对账和重建
|
||
type: scenario
|
||
domain: cross
|
||
stack: [data-pipeline, postgresql, mongodb, elasticsearch, vector-db, mq]
|
||
tags: [数据同步, outbox, CDC, 变更流, 幂等, 顺序, 对账, 全量重建]
|
||
level: advanced
|
||
prerequisites: [backend.data-pipeline.index, backend.mq.index]
|
||
est_hours: 3
|
||
related: [backend.elasticsearch.scenario.kb-search-sync, backend.data-pipeline.scenario.third-party-sync, backend.mongodb.scenario.conversation-store, backend.mq.reliability]
|
||
week: [6, 14]
|
||
---
|
||
|
||
# 数据同步：业务库到搜索引擎 / 向量库 / 报表库，outbox 与 CDC、幂等、顺序、对账和重建
|
||
|
||
## 业务背景与问题
|
||
|
||
业务数据在 PostgreSQL（订单、知识库文档）和 MongoDB（对话记录），需要同步到：ES（全文检索）、向量库（RAG）、报表库（报表 Agent）。现状是业务代码里“写完数据库顺手写 ES”：
|
||
|
||
1. 写 ES 失败时数据库已提交，两边不一致；
|
||
2. 同一条数据快速修改两次，ES 中留下的是旧版本；
|
||
3. 删除没有同步，搜索出已下架的商品；
|
||
4. 修改映射或换嵌入模型时不知道怎么重建。
|
||
|
||
## 约束
|
||
|
||
- 允许秒级到分钟级的延迟。
|
||
- 不能明显增加业务写入的耗时。
|
||
- 下游可以从源头完全重建（源头是唯一可信的数据）。
|
||
|
||
## 方案设计
|
||
|
||
1. **捕获变更**（二选一或组合）：
|
||
- **outbox 模式**：业务事务中同时写 outbox 表（事件类型、实体 id、版本号），后台进程投递到消息队列 → 保证“数据库提交 ⇔ 有事件”；
|
||
- **CDC**：读取数据库日志（PostgreSQL 逻辑复制 / Debezium，MongoDB 变更流），业务代码零侵入。
|
||
2. **消费与写入**：
|
||
- 消息只带 id 和版本，消费者回源读取最新数据再写下游（避免消息里的旧快照覆盖新数据）；
|
||
- 下游文档 id = 业务主键，写入带外部版本号，旧版本被拒绝；
|
||
- 删除事件 → 删除下游文档（或标记删除）；
|
||
- 同一实体的事件按实体 id 分区，保证顺序；
|
||
- 失败重试、死信、告警。
|
||
3. **对账**：每天比较源头与下游的数量和版本（按分片哈希），差异自动修复。
|
||
4. **全量重建**：新建索引 / 集合 → 全量导入（记录开始时间）→ 回放该时间之后的增量 → 校验 → 切换别名 → 保留旧索引回滚。
|
||
5. **监控**：同步延迟（事件产生到写入下游的时间）、积压量、失败数、对账差异。
|
||
|
||
## 关键代码
|
||
|
||
outbox（同一事务中写业务数据和事件）：
|
||
|
||
```sql
|
||
BEGIN;
|
||
UPDATE documents SET body = $2, version = version + 1, updated_at = now() WHERE id = $1;
|
||
INSERT INTO outbox (aggregate, aggregate_id, event, version)
|
||
SELECT 'document', id, 'updated', version FROM documents WHERE id = $1;
|
||
COMMIT;
|
||
```
|
||
|
||
消费者：回源 + 外部版本号（Elasticsearch Python 客户端，示意）：
|
||
|
||
```python
|
||
def handle(event: dict) -> None:
|
||
doc = repo.get(event["aggregate_id"]) # 回源读取最新数据
|
||
if doc is None or doc.deleted:
|
||
es.options(ignore_status=404).delete(index="kb", id=event["aggregate_id"])
|
||
return
|
||
es.index(index="kb", id=doc.id, document=to_search_doc(doc),
|
||
version=doc.version, version_type="external") # 版本小于等于已有版本时 ES 返回冲突，直接忽略
|
||
```
|
||
|
||
基础组件的可运行示例：[增量同步管道](../topics/backend/data-pipeline/examples/incremental-sync/)、[MongoDB 变更流](../topics/backend/mongodb/examples/conversations/)、[ES 别名切换](../topics/backend/elasticsearch/examples/hybrid-search/)。
|
||
|
||
## 取舍
|
||
|
||
| 决策 | outbox | CDC |
|
||
| --- | --- | --- |
|
||
| 侵入性 | 需要改业务代码 | 业务零侵入 |
|
||
| 事件语义 | 业务事件，含义清晰 | 行级变更，需要转换 |
|
||
| 运维 | 简单 | 需要部署和维护 CDC 组件、处理日志保留 |
|
||
| 适合 | 中小团队、事件少 | 表多、多个下游 |
|
||
|
||
## 易错点
|
||
|
||
- 业务代码中双写（数据库 + ES），没有事务保证。
|
||
- 消息中带完整快照，乱序时旧快照覆盖新数据。
|
||
- 忘记同步删除和权限变更。
|
||
- 重建期间的增量没有回放，切换后丢数据。
|
||
- 没有同步延迟监控，下游落后几小时才发现。
|
||
|
||
## 面试怎么讲
|
||
|
||
> “我们把‘写库顺手写 ES’改成了 outbox：业务事务里同时写 outbox 表，后台投递到队列，保证提交和事件一致。消费者只拿 id 回源读最新数据，用业务主键作为文档 id、版本号作为外部版本，乱序和重复都不会让旧数据覆盖新数据；删除也同步。每天按分片哈希对账；换映射或嵌入模型时新建索引全量导入、回放增量、校验后切别名。监控同步延迟和积压。”
|