MongoDB 实时同步到 Elasticsearch：用 ElasticRelay 5 分钟跑通（全量 + 增量 + 断点续传）

linjifan

1173人浏览 · 2025-12-17 11:01:28

linjifan · 2025-12-17 11:01:28 发布

当你已经用 MongoDB 承载业务数据，但又希望具备 Elasticsearch 的全文检索、聚合分析与近实时查询能力时，最经典的方案就是：

业务写 MongoDB
同步链路把变更实时投递到 Elasticsearch
查询侧走 Elasticsearch（搜索/统计/多维筛选）

很多人一开始会想到：Kafka + Debezium、Logstash、或者自己写同步服务。但它们要么架构偏重，要么运维成本高，要么“断点续传/一致性/重试/死信队列”很难一次做对。

ElasticRelay 的思路是：单一二进制 + 纯配置驱动 + Go 低资源占用，同时把 CDC（Change Data Capture）常见的工程问题（快照/增量/Checkpoint/DLQ）都内置掉，让你用更轻的方式把链路跑起来。

1. ElasticRelay 的 MongoDB → ES 同步是怎么工作的

核心链路可以理解为四步：

初始快照（可选）：启动时先把目标集合做一次全量扫表（snapshot），把现有数据写入 ES。
增量同步（Change Streams）：开启 MongoDB Change Streams 监听 insert/update/replace/delete。
Resume Token Checkpoint：每条变更事件都会带一个 resume token；ElasticRelay 会把它落盘（默认 mongodb_checkpoints.json），用于断点续传。
ES Bulk 写入：ElasticRelay 将事件转换成 JSON，按批次 bulk 写入 Elasticsearch。

索引命名规则：

默认：<index_prefix>-<collection>（collection 会转小写）
MongoDB 事件里集合名字段：_collection
快照记录里集合名字段：_table（ES sink 同时支持 _table 与 _collection）

2. 前置条件（别跳过）

2.1 MongoDB 必须是 Replica Set 或 Sharded Cluster

Change Streams 的硬要求：

MongoDB 4.0+（推荐 6/7）
Replica Set 或 Sharded Cluster
单机 standalone 不支持 Change Streams（会报 “Change Stream not supported”）

2.2 MongoDB 用户权限（最小化建议）

Change Streams 通常需要：

业务库的 readWrite（至少 read）
local 库的 read（用于 Change Streams/相关内部读取）

仓库自带的 Docker 初始化脚本已经帮你创建了用户：

用户：elasticrelay_user
密码：elasticrelay_pass

2.3 Elasticsearch

ES 7.x / 8.x 均可
如果开启了安全认证，准备好 user/password

3. 5 分钟跑通（Docker 起 MongoDB + ES，本机跑 ElasticRelay）

下面这套流程尽量复用仓库现成的 compose 与配置文件。

3.1 启动 MongoDB Replica Set（仓库已内置）

在项目根目录执行：

docker-compose up -d mongodb mongodb-init

它会拉起：

elasticrelay-mongodb：MongoDB 7.0（副本集 rs0）
elasticrelay-mongodb-init：一次性初始化副本集与测试数据

3.2 启动 Elasticsearch

仓库的 docker-compose.yml 里 Elasticsearch 默认是注释掉的。你可以选择：

方式 A：取消 docker-compose.yml 里 elasticsearch 服务的注释，然后执行：

docker-compose up -d elasticsearch

方式 B：使用你已有的 Elasticsearch（只要 config/mongodb_config.json 里 addresses 指向你的 ES 即可）。

3.3 构建 ElasticRelay

chmod +x scripts/build.sh
./scripts/build.sh

产物默认在 bin/elasticrelay。

3.4 配置 MongoDB → ES 同步任务

你可以直接使用仓库自带示例：config/mongodb_config.json。

其中最关键的四块配置：

data_sources：MongoDB 连接信息、监听哪些集合（table_filters）
sinks：Elasticsearch 地址与索引前缀（index_prefix）
jobs：把 source 与 sink 绑起来，并配置是否先做全量（initial_sync）
global：日志、metrics、DLQ 等

最小可跑版本示意（与仓库示例一致）：

{
  "version": "2.0",
  "data_sources": [
    {
      "id": "mongodb-primary",
      "type": "mongodb",
      "host": "localhost",
      "port": 27017,
      "user": "elasticrelay_user",
      "password": "elasticrelay_pass",
      "database": "elasticrelay",
      "table_filters": ["users", "orders", "products"]
    }
  ],
  "sinks": [
    {
      "id": "es-primary",
      "type": "elasticsearch",
      "addresses": ["http://localhost:9200"],
      "options": { "index_prefix": "elasticrelay_mongo" }
    }
  ],
  "jobs": [
    {
      "id": "mongodb-to-es",
      "source_id": "mongodb-primary",
      "sink_id": "es-primary",
      "enabled": true,
      "options": { "initial_sync": true }
    }
  ],
  "global": { "log_level": "info" }
}

3.5 启动 ElasticRelay

./bin/elasticrelay --config config/mongodb_config.json

启动后你会看到类似日志：

连接 MongoDB 成功
开始快照（如果 initial_sync=true）
Change Streams started
写入 ES（自动创建索引）

4. 验证同步：插入/更新/删除都能实时反映到 ES

4.1 查看 ES 索引是否生成

如果 index_prefix=elasticrelay_mongo，同步 users 集合会生成索引：

elasticrelay_mongo-users

执行：

curl -X GET "http://localhost:9200/_cat/indices?v"

4.2 查一条数据

curl -X GET "http://localhost:9200/elasticrelay_mongo-users/_search?pretty" \
  -H 'Content-Type: application/json' \
  -d '{"query":{"match_all":{}},"size":10}'

4.3 触发增量：对 MongoDB 做写入

如果你用的是仓库自带 Docker MongoDB：

docker exec -it elasticrelay-mongodb mongosh \
  -u elasticrelay_user -p elasticrelay_pass \
  --authenticationDatabase admin

然后执行：

use elasticrelay

// insert
 db.users.insertOne({
  name: 'Test User',
  email: 'test@example.com',
  age: 30,
  created_at: new Date(),
  updated_at: new Date()
})

// update
 db.users.updateOne({email:'test@example.com'}, {$set:{age: 31, updated_at: new Date()}})

// delete
 db.users.deleteOne({email:'test@example.com'})

回到 ES 再查一次，你应该能看到变化实时出现（或删除生效）。

5. 运维要点：Checkpoint、重启恢复、DLQ

5.1 Checkpoint（断点续传）

MongoDB CDC 使用 Change Streams 的 resume token 做 checkpoint。

默认落盘文件：mongodb_checkpoints.json
进程重启：会先尝试加载该文件，从上次位置继续

常见注意事项：

如果 resume token 因 oplog 窗口太短而失效，会出现类似 “Resume token not found / resume of change stream was not possible”。这时需要：
- 增大 oplog window（生产推荐）
- 或者清理 checkpoint 重新从当前/全量开始（谨慎）