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原子操作。每条事件到达消费者后只做一次主键 upsertconst 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 查询端优化把复杂聚合留在“最后一公里”预聚合层做好之后查询端基本是点查或小范围聚合。比如看某渠道某分钟的 PVdb.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 症状、原因、对策速查表症状可能原因处理方式聚合查询偶尔报内存超过 100MB管道里有大$sort或$group且没走索引给$match前的字段加复合索引或给$sort加allowDiskUse: trueChange Streams 消费者断连闲置时间过长连接被服务端回收用tryNext()代替hasNext()定期发心跳预聚合upsert偶发重复主键异常同一条 key 的并发upsert捕获E11000重复键错误改成updateOne重试数据量不大但查询变慢过滤条件没有完全命中复合索引回表严重用explain(executionStats)看totalDocsExamined是否远大于nReturned某个分片写入特别多分片键选成了低基数字段改成{ channel: 1, ts: 1 }或 hashed 分片键TTL 集合不删数据TTL 索引建错了字段或字段不是日期类型确认索引字段是日期类型且位于文档顶层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。
