利用 Oplog 实现高性能数据同步管道与增量ETL

讲解如何直接读取 Oplog 实现自定义的数据复制、实时分析管道，以及注意事项。 · 难度：入门 · +10XP

利用 Oplog 实现高性能数据同步管道与增量ETL

MongoDB 的 Oplog（操作日志）记录了所有写入操作，是实现数据复制的基础。本教程将绕过 Change Streams，直接讲解如何高效扫描 Oplog 实现低延迟的增量数据同步管道（例如同步到 Elasticsearch 或者 Redis）。你会学习 Oplog 的文档结构（ts、op、ns、o、o2 字段）、如何通过 ts 游标断点续传，以及如何过滤特定集合与操作类型（仅 insert/update/delete）。还会讨论 Oplog 的保留窗口、主备切换时的令牌稳定性，以及使用 Oplog 做跨版本数据迁移的技巧。注意：直接读取 Oplog 需要足够的权限并小心处理性能影响。

// 连接到 local 数据库读取 oplog.rs 集合
const oplog = client.db('local').collection('oplog.rs');
const cursor = oplog.find({
  ns: /^mydb\.users$/,
  ts: { $gt: lastTimestamp },
  op: { $in: ['i', 'u', 'd'] }
}).sort({ $natural: 1 }).addCursorFlag('noCursorTimeout', true);
while (await cursor.hasNext()) {
  const entry = await cursor.next();
  // 解析 entry.o 字段获取变更数据
  console.log(entry.op, entry.ns, entry.o);
}