本周处理了一个亿级数据量的生产任务：将一亿条原始流水导入数据库，并针对 2000 万会员进行等级重算与回写。这次任务的核心难点不在于逻辑实现，而在于如何在高并发写入下维持 Oplog Window（同步窗口） 的平衡，以及如何利用 $merge 替代传统的更新方式。

1. 原始数据导入阶段的 Oplog 压力

在大规模数据迁移中，海量数据的初始导入是对存储引擎的第一轮冲击。导入这一亿条流水时，瞬时写入量会导致 Oplog 的生成速度极快，从而引发 Oplog Window 的急剧下降。

什么是 Oplog Window？为什么它下降很危险？

Oplog（Operations Log）是 MongoDB 副本集同步的核心。它是一个固定大小的循环文件（Capped Collection），记录了主库的所有写操作。Oplog Window 指的是当前 Oplog 日志能够覆盖的时间跨度。

• 危险所在：如果 Oplog Window 下降到 10 分钟，意味着从库必须在 10 分钟内完成数据同步。一旦从库因为网络抖动或 IO 繁忙导致延迟超过 10 分钟，主库上旧的 Oplog 就会被新日志覆盖。
• 后果：一旦 Oplog 被覆盖，从库将永远无法通过增量同步来追平主库，此时从库会进入 RECOVERING 状态，最终只能通过耗时巨大的全量同步（Initial Sync）来修复，这在生产环境下是极其危险的。

Oplog 占用估算逻辑： 预计 Oplog 空间 ≈ 迁移记录数 × 记录平均大小 × 1.2（系统开销系数）

以本次任务为例，近一亿条原始流水，如果单条记录约 200 Bytes，导入过程产生的 Oplog 就会接近 20GB。

优化手段：

• 临时扩容：通过以下命令在线调大 Oplog 空间。注意 size 的单位是 MB，这里我们将其设置为 30GB（即 30720MB）：

db.adminCommand({ replSetResizeOplog: 1, size: 30720 });

• 速率调控：实时监控 replication lag，根据从库的同步延迟动态调整导入线程数。

---

2. $merge：替代传统 Update 的工业级方案

在完成流水聚合后，我们需要对 700 多万发生变动的会员进行数据回写。相比传统的循环 update，我采用了 $merge 管道。

// 使用 $merge 实现高效的原地（In-place）回写
db.member_growth_temp.aggregate([
  { 
    $group: { 
      _id: "$memberId", 
      totalGrowth: { $sum: "$growth" } 
    } 
  },
  {
    $match: {
      totalGrowth: { $gt: 0 } // 过滤有效数据
    }
  },
  {
    $merge: {
      into: "member",
      on: "_id",
      whenMatched: "merge", // 仅合并变更字段，极大降低写放大
      whenNotMatched: "discard" 
    }
  }
], { allowDiskUse: true });

$merge 的底层核心优势：

• 消除网络往返：计算结果直接在数据库内部流入目标集合，不需要经过应用层分发，避免了海量的序列化开销。
• 降低 Oplog 负载：通过 whenMatched: "merge"，我们只针对变动字段进行 Partial Update。由于这阶段回写的数据量远小于第一阶段的导入，针对 700 万条记录的精准更新对 Oplog 的压力明显缓解。

---

3. 进程监控与锁状态观察

在大规模 $merge 执行期间，你会发现 currentOp 视图中的进程可能会在写入阶段变得“难以捕捉”。

此时不应盲目判断任务挂起，而应观察服务器的 Lock Acquisition（锁获取） 指标。当看到锁获取计数器保持高频跳动，说明底层正在以微秒级的频率进行批次提交。这种异步批处理机制正是 $merge 能够快速处理千万级更新的原因。

---

结语

仰赖于 $merge 的超高效率，最终停机后，实际数据迁移更新占用的事件只有 30 分钟，最终是有惊无险，如预期一样完成了。