Mongo实现数据持续聚合汇总的方案咨询

嘿Ian，你的实时交易数据聚合场景在分布式系统里很典型，我来给你分享几个能解决这些痛点的方案：

一、解决$out覆盖目标集合的问题：改用$merge阶段 #

从MongoDB 4.2版本开始，$merge就是$out的升级版，完美解决了覆盖和分片写入的问题：

• 核心优势：支持向目标集合追加数据，而非直接覆盖；同时允许写入分片集合，刚好匹配你的分片存储场景。
• 配置思路：
  1. 在聚合管道的末尾添加$addFields，给每个聚合结果打上唯一标识（比如聚合的时间窗口ID，例如"2024-05-20T14:30:00/5s"），用来区分不同批次的聚合数据。
  2. 使用$merge替代$out，配置匹配规则和冲突处理逻辑：

     {
       $merge: {
         into: "aggregated_transactions", // 目标集合名称
         on: "window_id", // 用来匹配已有文档的唯一标识字段
         whenMatched: "keepExisting", // 若存在相同标识的文档，保留原数据（可按需选"replace"/"merge"）
         whenNotMatched: "insert" // 无匹配时插入新文档，实现追加效果
       }
     }
     

  这样每次聚合的结果都会作为新文档追加到目标集合，不会覆盖之前的数据。

二、优化已处理数据的追踪与事务性问题 #

你当前标记+删除的方案没有事务保障，确实容易出现数据丢失或重复处理的情况，这里有几个更可靠的思路：

1. 基于时间窗口的无侵入式追踪 #

如果你的交易数据带有准确的createdAt时间戳，这是最简单的方案：

• 每次聚合任务执行时，只处理上一次聚合结束时间到当前时间的交易数据，同时将本次聚合的结束时间记录在一个单独的配置集合（比如aggregation_config）中，存储字段last_processed_time。
• 比如每5秒执行一次，就处理createdAt在[last_time, last_time+5s)区间内的文档，处理完成后更新last_processed_time为当前时间。
• 优势：不需要修改原交易数据，也不需要删除操作；如果某次聚合失败，下次可以重新处理同一个时间窗口的数据，不会丢失；分片集合下可以通过createdAt（如果是分片键的一部分）快速定位分片，提升聚合性能。
• 注意：如果存在延迟到达的交易数据，可以适当放宽时间窗口（比如每次处理前10秒的数据，跳过已处理的部分），或者定期补跑历史窗口。

2. 利用变更流（Change Streams）实现实时追踪 #

如果你的集群是副本集或分片集群（你的场景是分片集合，刚好满足），可以用变更流监听交易集合的插入事件：

• 维护一个内存缓冲区，收集最近5秒内的新插入文档；
• 每5秒触发一次聚合，将缓冲区的数据聚合后写入目标集合；
• 变更流的游标会自动记录最后处理的位置，即使服务重启，也能从上次中断的地方继续，不会重复处理或丢失数据。
• 优势：完全不需要标记或修改原数据，天然支持断点续传；适合实时性要求高的场景。

3. 事务+状态标记的可靠处理 #

如果必须对原交易数据做标记，建议结合MongoDB事务来保障原子性：

• 步骤如下（需要集群支持事务）：
  1. 开启一个事务；
  2. 批量查询status: "unprocessed"的交易文档，将其更新为status: "processing"（用批量更新操作，注意加上查询条件限制范围，避免全表扫描）；
  3. 对这些标记为processing的文档执行聚合；
  4. 将聚合结果写入目标集合；
  5. 将原交易文档的status改为processed（或者直接删除，根据你的数据保留策略）；
  6. 提交事务。
• 优势：如果聚合过程中出现失败，事务会自动回滚，原文档的状态会恢复为unprocessed，下次可以重新处理，避免数据丢失；全程原子性，不会出现半处理的状态。
• 注意：分片集合下的事务需要确保查询条件命中分片键，否则会触发跨分片事务，性能可能受影响。

额外的性能优化建议 #

• 对于分片集合的聚合，尽量在聚合管道的早期阶段添加分片键过滤，减少跨分片的数据传输；
• 如果聚合数据量较大，记得开启allowDiskUse: true，避免内存不足的问题；
• 时间窗口聚合可以结合$bucket或$bucketAuto阶段，更高效地按时间分桶处理。

内容的提问来源于stack exchange，提问作者IanG