基于MongoDB的实时分析系统设计：数据模型与预聚合实践

1. 先把需求拆透：实时分析到底要“实时”到什么程度

每次有人跟我说“要做一个实时分析系统”，我第一反应不是问用什么数据库，而是先问：这个“实时”是多实时？是要秒级报警，还是分钟级大屏刷新？两者的架构差别非常大。

我最近帮某模拟项目 X 做的那套点击流分析，业务方一开始的说法是“数据进来 1 秒内要出结果”。后来把需求拆开看，真正要求秒级的是支付异常监控，而常见的 PV/UV、渠道转化、留存漏斗，5 分钟延迟完全可接受。如果所有指标都按秒级标准设计，成本至少翻三倍。

做实时分析系统，最容易犯的错就是“全部指标一把抓”。这个坑我踩过不止一次，所以现在拿到需求的第一件事，永远是做实时性分级。

1.1 实时性分级是第一个要定的事

我把实时需求分成三档，每一档对应的技术选型和投入完全不一样：

• 秒级（1~5 秒）：异常报警、风控拦截、在线推荐。这类需要真正的流式处理链路，事件写入后要立刻触发计算，不能靠定时任务。
• 分钟级（1~5 分钟）：大屏指标、运营看板、渠道监控。这类可以接受微批量，每 1~2 分钟把增量数据扫一遍，用定时触发的聚合管道就够了。
• 小时级及以上：趋势分析、深挖报表。这一类其实已经不能算实时了，但对存储的容量规划影响很大，往往要和实时数据分开存储。

每一档对存储和计算的要求完全不同。秒级的核心是延迟，分钟级的核心是吞吐，小时级的核心是成本。你在物理上也没办法让一个 MongoDB 集群既扛住百万级写入，又对每一条做复杂的多维聚合，还要在 1 秒内返回。这不现实。

所以设计实时分析系统的第一步，就是把所有指标分类，明确每类指标的延迟上限。这一步看起来不性感，但直接决定后面的数据模型、索引策略和管道设计。如果需求方说“都要实时”，你就把成本算给他看：秒级指标占用多少写入量、多少聚合计算量、多少存储资源，让他自己排优先级。

1.2 为什么最后是 MongoDB，不是 ClickHouse 或 Elasticsearch

这套系统当时也做了几轮选型。ClickHouse 在聚合查询上确实快，但它更适合“数据基本定型、字段少变动”的场景。我们这边事件结构经常要加字段，比如运营临时要埋一个“来源页面层级”，用 ClickHouse 就要改表结构，改完还要考虑数据回填。Elasticsearch 适合全文检索和日志分析，做精确去重和嵌套聚合要小心磁盘和内存，索引膨胀的问题在写入量上来之后会让人很头疼。

MongoDB 的优势在两个地方：一是文档模型足够灵活，事件字段随时可以加，不需要预先定义严格的 schema；二是聚合管线的能力覆盖了绝大多数分析场景，从 $match、$group 到 $bucket、$merge 都有，配合 Change Streams 能直接把增量数据接进实时计算链路。对于一个需要快速迭代、数据模型还没完全定死的系统来说，这个优势非常实用。

MongoDB 不适合做的是那种超大宽表的深度分析，比如一次性 join 十几张表、跑几十亿行数据的复杂报表。那是数据仓库的活。实时分析系统讲究的是“能快速看到变化趋势”，不是“做全量离线计算”，所以 MongoDB 在这个场景里扮演的角色是：承接实时写入、做增量预聚合、支撑低延迟查询。搞清楚这个定位，后面很多设计就不会走偏。

2. 数据模型：别按关系库思路设计文档

很多人用 MongoDB 做分析系统，第一反应是建一张“宽表”，把所有字段塞进去，然后靠索引去扛查询。如果数据量小没问题，一旦上了千万量级，你会发现索引再全也扛不住无节制的聚合扫描，而且写入性能会被索引拖垮。

我的做法是建两层：明细层和预聚合层。明细层负责原始事件，保留尽可能多的字段；预聚合层负责按时间和维度做汇总，承担绝大多数读请求。两层的数据量和访问模式完全不同，索引策略也要分开设计。

2.1 事件明细集合：保留原始事实

明细集合负责承接原始数据，字段设计以“不过度加工”为原则。比如我们处理的一条浏览事件文档是这样的：

{ _id: ObjectId("66f8c2d1a1b2c3d4e5f6a7b8"), event_type: "view", user_id: "u_1024", session_id: "s_7788", item_id: "p_332", channel: "app", ts: ISODate("2025-01-01T10:00:08.123Z"), extra: { page_level: 2, referer: "home", device: "ios" } }

extra是一个自由对象，用来放那些“可能之后会用到但又不确定”的字段。这样埋点上线初期不需要把所有字段都固化成顶层字段，后面确认哪个字段高频查询，再提出来建独立字段和索引。

这里我踩过的一个坑是：不要一上来就建一堆单字段索引。MongoDB 的查询优化器对单字段索引和复合索引的选择逻辑很直接，如果某个查询的过滤条件横跨多个字段，单字段索引只能命中其中一个，其余过滤条件全在回表之后处理，性能立刻崩。更好的做法是先分析查询模式，再加两三个覆盖核心过滤维度的复合索引。

2.2 预聚合集合：用桶来换时间

实时分析的性能瓶颈其实不在写，而在读。查询端要的是秒级返回，但明细数据往往有几千万条，每次现算很吃力，而且并发查询一多，CPU 和内存都会被拖垮。

我的方案是按“时间桶 + 维度键”做预聚合。比如一条分钟级聚合文档：

{ _id: "2025-01-01T10:00:00Z|view|app", bucket_key: "2025-01-01T10:00:00Z", event_type: "view", channel: "app", count: 1520, unique_users: 847, sum_amount: 3890.5, last_ts: ISODate("2025-01-01T10:00:59Z") }

_id直接由“分钟 + 事件类型 + 渠道”拼成，这样upsert时可以按_id查询，走主键索引，非常快。后面查渠道转化、时段对比就都在这个集合上做，数据量从千万降到几十万，性能自然上了一个台阶。

时间桶的粒度不要一刀切。我通常建三套：

• 分钟级桶：给秒级和分钟级看板用，保留 1 天。
• 小时级桶：给趋势分析用，保留 30 天。
• 天级桶：给长期对比用，保留更长时间，可以归档到低成本存储。

TTL 索引直接设在这些桶集合的固定日期字段上，到时间自动淘汰，连清理脚本都省了。这个设计对运维来说非常友好，不需要额外维护一批定时删除任务。

2.3 分片键：别选会造成热点的字段

当单集合扛不住写入时，就要考虑分片。分片键的选择，是 MongoDB 实时分析系统里风险最高的一个决策，比索引设计还要防不胜防。

我见过最典型的问题是用channel做分片键。结果某一个渠道搞活动，全部流量打在这个字段的一个值上，正好那个 chunk 被命中，整个集群出现一个热点分片，其他分片闲着没事干。更难受的是，这种热点不是固定的，活动一结束热点又转移了，你必须不停调整分片策略。

正确的做法是把时间字段加进分片键，比如{ channel: 1, ts: 1 }，让流量在时间维度上自然散开。或者干脆用 hashed 分片键，虽然范围查询会差一点，但写入均衡性最好。如果你拿不准，可以用 hashed 保底，后面再根据实际访问模式调优。

这里要补充一个经验：不要在一开始就分片。先单节点或复制集跑起来，观察写入吞吐和 CPU 使用率，如果出现持续打满，再考虑分片。提前分片只会增加运维复杂度，而且 MongoDB 的分片集群在查询路由上会有额外开销，小数据量时反而更慢。

3. 聚合管线与实时计算改造

数据模型定好，接下来是整个系统的核心：怎么把新进来的数据变成预聚合结果，并且保证实时性。

这里要强调的是，实时分析系统的“计算”，不是等数据落库后再去跑一堆离线作业，而是尽量把计算前置，让每个增量事件都触发出一个很小的聚合更新。流式处理的核心思路，说穿了就是“增量代替全量”。

3.1 用 Change Streams 监听增量事件

MongoDB 提供的 Change Streams 是天然的实时入口。它可以监听集合的写入、更新、删除事件，而且支持管道式过滤。比轮询ts字段或者用oplog自己拼方案，要省事得多。

一个实时消费者大概长这样：

const pipeline = [ { $match: { operationType: "insert", "fullDocument.event_type": { $in: ["view", "purchase"] } } }, { $project: { doc: "$fullDocument" } } ]; const cursor = db.events.watch(pipeline, { fullDocument: "updateLookup" }); while (await cursor.hasNext()) { const change = await cursor.next(); const doc = change.doc; // 在这里做增量预聚合 }

这里有几个关键点。fullDocument: "updateLookup"是为了在 update 事件里拿到整条文档，如果没有它，update 事件只返回变更字段，很多后处理逻辑就没法做。监听操作要放在一个持续运行的进程中，不能用“跑一次就退”的脚本，否则resume token维护会很麻烦，每次重启都得重新消费一大段历史事件。

如果是多分片集群，Change Streams 会把所有分片的事件合并成一个流，但跨分片的全局顺序无法保证。所以不要在消费者里依赖跨事件的顺序，尽量让每条事件的处理是幂等的。这个约束对预聚合逻辑来说问题不大，因为$inc本身就是幂等的累加操作。

3.2 预聚合的两种实现方式

第一种是用$merge在 MongoDB 内部直接做聚合，适合分钟级批量处理。每 1~2 分钟跑一次增量扫描，把新到的事件分组后合并进预聚合集合：

db.events.aggregate([ { $match: { ts: { $gte: startTime }, event_type: { $in: ["view", "purchase"] } } }, { $group: { _id: { bucket: { $dateTrunc: { date: "$ts", unit: "minute" } }, event_type: "$event_type", channel: "$channel" }, count: { $sum: 1 }, unique_users: { $addToSet: "$user_id" } } }, { $merge: { into: "minute_stats", on: "_id", whenMatched: "replace", whenNotMatched: "insert" } } ])

这种方案优点是简单，不用额外写消费者进程。缺点是如果事件量很大，每轮扫描都要走一遍索引和分组，MongoDB 的 100MB 内存排序限制会成为瓶颈。而且$addToSet在量大之后很占内存，用户数一多，跑了没几天管道就会报内存超限。

第二种是代码里做增量更新，配合$inc原子操作。每条事件到达消费者后，只做一次主键 upsert：

const bucketStart = new Date(Math.floor(doc.ts.getTime() / 60000) * 60000); const key = `${bucketStart.toISOString()}|${doc.event_type}|${doc.channel}`; db.minute_stats.updateOne( { _id: key }, { $inc: { count: 1 }, $set: { last_ts: doc.ts, updated_at: new Date() } }, { upsert: true } );

这种看起来简单，但很稳。写路径非常短，每条事件只更新一个主键文档，没有排序、没有分组、没有临时文件。唯一要留意的是并发 upsert 时可能会遇到重复主键的异常，需要捕获E11000重复键错误，然后在 catch 里做一次普通updateOne兜底。

我个人的建议很明确：秒级场景用第二种，分钟级场景用$merge。不要在 Change Streams 消费者里用$addToSet做精确去重，那样内存会被瞬间打爆。精确去重指标要么用 HyperLogLog 之类的近似算法，要么放到查询端按预聚合结果再算，不要试图在实时链路上维护全集。

3.3 查询端优化：把复杂聚合留在“最后一公里”

预聚合层做好之后，查询端基本是点查或小范围聚合。比如看某渠道某分钟的 PV：

db.minute_stats.aggregate([ { $match: { bucket_key: { $gte: start, $lte: end }, channel: "app", event_type: "view" } }, { $group: { _id: "$channel", total: { $sum: "$count" } } } ])

如果发现某个查询经常出现“全集扫描”，要回头检查$match能否命中索引。比如上面的查询，应该在minute_stats上建复合索引{ bucket_key: 1, event_type: 1, channel: 1 }，这样 MongoDB 只需要扫描一个很小的数据范围。

还有一个常见误区：不要把明细层的索引完全照搬到预聚合层。预聚合集合的数据量小很多，索引数量要克制，保留 2~3 个复合索引就足够了。索引越多，写入性能越差，这在预聚合高频更新时非常明显。你每加一个索引，每次updateOne就多维护一棵 B 树，积少成多，写性能会肉眼可见地往下掉。

查询端还有一个小技巧：把大屏类的查询结果做缓存。比如一个看板每 5 秒刷新一次，如果每次都去读 MongoDB，其实大部分结果都一样。可以缓存 5 秒，命中缓存就直接返回，只有超过 5 秒才真正查库。这个优化能在不改变架构的情况下，把集群的读压力降下一大截。

4. 常见问题与排查经验

只写成功路径不给踩坑经验，等于没写。我把实际运行中遇到的几个典型问题和排查思路整理一下，都是一些常规文档里不会细讲的细节。

4.1 症状、原因、对策速查表

4.2 一个真实的索引踩坑记录

这套系统上线时，我发现events集合的view事件查询经常扫描几十万条文档，但返回只有几百条。explain结果显示，查询走了{ ts: -1 }单字段索引，后面所有event_type和channel过滤都在回表后处理。

原因是复合索引建成了{ event_type: 1, channel: 1, ts: -1 }。看起来没问题，但查询里ts是范围条件，event_type和channel如果没有作为等值条件出现，索引最左前缀的策略会引发意想不到的行为，导致范围扫描没走对。改成{ ts: -1, event_type: 1, channel: 1 }后，ts范围定位能直接走索引，速度一下子从 800ms 降到 20ms 左右。

这说明复合索引的字段顺序，必须结合查询条件里的字段类型来定。范围字段放在最前面，等值字段放在后面，是比较稳妥的策略。数据量一大，这个细节会放大成灾难。

4.3 变更流消费的取舍

我试过把 Change Streams 消费逻辑直接塞到业务接口里，也就是边写入边聚合。结果写入量一高，接口整个被拖慢，用户侧的响应时间暴增到 2 秒多。后来把消费逻辑拆到独立进程中，写入接口只负责insertOne，消费者在后面慢慢处理。实时性依然能满足秒级要求，但系统的稳定性和扩展性好太多。

如果团队里已经有消息队列，也可以把 MongoDB 作为上游，通过 Change Streams 把事件推给消息队列，再由流处理引擎做预聚合，最后写回 MongoDB。这套链路听起来更复杂，但好处是削峰和消息回溯能力。不过我们这套系统没有那么重的依赖，直接用 MongoDB 驱动写消费者就够了。每个系统都有自己的复杂度预算，能少引入一套组件就少引入一套。

还有一个要注意的点：Change Streams 的resume token要持久化。如果消费者进程重启后从头开始消费，会把全量历史事件重新刷一遍，预聚合集合瞬间就会崩。正确的做法是把 token 存到数据库或者配置中心，重启后接着上次的位置续传。

5. 个人体会与一点后续扩展

踩过几次坑之后，我现在的原则是：MongoDB 做实时分析，核心不是数据库本身有多快，而是数据模型和预聚合策略。明细层负责“存得下”，预聚合层负责“查得快”，Change Streams 负责“接得住”，三者分工明确，系统才能真正稳定运行。

后续如果要扩展，我觉得可以从两个方向入手。一是把明细层的事件归档到更便宜的存储，比如对象存储或者专门的冷数据集合，减少热数据量。二是往预聚合层加入更丰富的维度组合，比如把用户画像里的年龄段、会员等级也放进桶键里，这样运营就能直接看不同人群的实时转化，不用再把明细拉出来重算。

如果你也在做类似的系统，可以先从“分钟级 + 事件类型 + 渠道”这个最小模型开始跑，确认稳定了，再往秒级和更多维度上扩。这套路径我用下来是比较顺的。最后再说一个细节：预聚合的间隔时间不要拍脑袋定，先看写入量，再算一下每分钟大概产生多少条事件，然后决定是每分钟跑一次还是每两分钟跑一次。数据量不大时，跑太勤反而浪费 CPU。