【SkyWalking从入门到精通】第45篇：OAP集群的计算流：两级聚合的优雅舞蹈

上一篇【第44篇】自定义OAL指标实战：打造你的专属可观测仪表盘
下一篇【第46篇】OAP集群通信协议：节点间对话的暗语

一、开篇：从一个极端的假设开始

假设你的系统每秒产生10,000条Span数据，部署了10个OAP节点组成集群。如果每个OAP节点都把原始Span数据写入存储，然后由存储层负责聚合，会发生什么？

10个OAP节点 × 10,000条/秒 = 100,000次/秒写入ES 再加上： 100,000次/秒 × (AVG + SUM + COUNT + P99) = 400,000次聚合查询/秒 结论：在写入的同时还要做聚合查询，ES会直接挂掉

这就是SkyWalking采用两级聚合的根本原因。它的设计思路就像物流体系的分级配送：

各门店订单 区域集散中心 全国总仓 ========= ========== ========= L1 本地聚合 L2 集群聚合 写入存储 (每个OAP节点) (集群协调) (ES/MySQL) 每收到一条数据 收集所有节点结果 持久化 立刻在内存聚合 做二次合并计算 供查询 图1：两级聚合 = 分布式计算中的"物流分拣中心"

二、两级聚合的架构全景

+-----------------------+ | Agent探针集群 | | (100+ Service, 1000+ | | Instances) | +-----------+-----------+ | gRPC (Segment/Log/Meter) | +---------------------------+---------------------------+ | | | v v v +----------+----------+ +----------+----------+ +----------+----------+ | OAP Node 1 | | OAP Node 2 | | OAP Node N | | | | | | | | [L1 本地聚合] | | [L1 本地聚合] | | [L1 本地聚合] | | | | | | | | 收到数据 -> 内存聚合 | | 收到数据 -> 内存聚合 | | 收到数据 -> 内存聚合 | | (分钟级窗口) | | (分钟级窗口) | | (分钟级窗口) | +----------+-----------+ +----------+-----------+ +----------+-----------+ | | | | L1结果 | L1结果 | L1结果 +---------------------------+---------------------------+ | +-----------v-----------+ | 集群协调层 | | (Nacos/ZK/K8s) | +-----------+-----------+ | +-----------v-----------+ | L2 全局聚合 | | | | 合并所有节点的L1结果 | | 生成最终的小时/天指标 | | 写入存储 | +-----------+-----------+ | +-----------v-----------+ | 存储层 | | (ES/MySQL/H2) | +-----------------------+ 图2：两级聚合的完整架构图

三、L1聚合：每个节点的"本地会计"

L1聚合是分布式的第一层防线。它运行在每个OAP节点内部，直接消费探针数据。

3.1 L1聚合的工作流程

/** * L1聚合的核心逻辑（简化版） */publicclassL1AggregationWorker{// 分钟级聚合窗口// Key: MetricKey(指标名+实体ID+时间窗口)// Value: 聚合中Metric对象privatefinalMap<MetricKey,Metrics>l1Buffer=newConcurrentHashMap<>();/** * 每收到一条探针数据的回调 */publicvoidonSegmentReceived(SegmentObjectsegment){// 1. 解析Segment，提取各维度数据List<Source>sources=segmentToSources(segment);// 2. 对每个Source，查找对应的OAL指标定义for(Sourcesource:sources){for(AnalyzeDefinitiondef:getOALDefinitions(source)){// 3. 确定时间窗口longminuteWindow=truncateToMinute(segment.getStartTime());// 4. 构建聚合KeyMetricKeykey=newMetricKey(def.getMetricName(),source.getEntityId(),minuteWindow);// 5. 获取或创建聚合对象Metricsexisting=l1Buffer.get(key);if(existing==null){existing=def.createMetrics();l1Buffer.put(key,existing);}// 6. 执行流式聚合（调用生成的combine方法）def.getProcessor().combine(source,existing,context);}}}/** * L1聚合窗口闭合时的回调（每分钟） */@Scheduled(fixedRate=60_000)publicvoidflushMinuteWindow(){longcurrentMinute=truncateToMinute(System.currentTimeMillis());// 找出所有上一分钟窗口的聚合结果longtargetMinute=currentMinute-60_000;List<Metrics>completedMetrics=l1Buffer.values().stream().filter(m->m.getTimeBucket()==targetMinute).collect(Collectors.toList());for(Metricsmetrics:completedMetrics){// 1. 写入内存中的小时聚合窗口（本地）mergeToHourWindow(metrics);// 2. 发送给L2聚合协调器（集群）l2AggregationCoordinator.send(metrics);// 3. 从L1缓存中移除（已处理完毕）l1Buffer.remove(MetricKey.from(metrics));}}}

3.2 L1聚合的关键特征

四、L2聚合：集群层面的"总会计师"

L2聚合汇总所有OAP节点的L1结果，计算集群级别的最终指标。

4.1 L2聚合的完整流程

/** * L2聚合协调器 * 运行在集群中的某个节点上（或所有节点协商分工） */publicclassL2AggregationCoordinator{// L2聚合缓冲区// 收集所有节点发来的L1结果privatefinalMap<MetricKey,List<Metrics>>l2CollectBuffer;// 分配策略：通过consistent hash决定哪个节点处理哪个指标privatefinalConsistentHash<String>partitioner;/** * 接收L1聚合结果 */publicvoidreceive(Metricsl1Result,StringsenderNodeId){MetricKeykey=MetricKey.from(l1Result);// 确定该指标由哪个节点负责L2聚合StringtargetNode=partitioner.getNode(key.toString());if(isSelf(targetNode)){// 由本节点处理L2聚合aggregateL2(key,l1Result);}else{// 转发给目标节点remoteClient.forward(targetNode,l1Result);}}/** * 执行L2聚合 */privatevoidaggregateL2(MetricKeykey,Metricsl1Result){// 获取当前L2窗口的聚合结果Metricsexisting=l2Buffer.get(key);if(existing==null){// 第一个到达的结果，直接记录l2Buffer.put(key,l1Result);}else{// 后续到达的结果，调用combine合并// L2版本的combine接受两个Metrics对象做合并MetricsDefinitiondef=metricsRegistry.getDefinition(key.getMetricName());def.getProcessor().combine(l1Result,existing);}// 检查是否收集了足够的结果（90%以上节点已上报）if(checkCollectionComplete(key)){MetricsfinalResult=l2Buffer.remove(key);// 写入存储storageWriter.write(finalResult);}}}

4.2 L2聚合的窗口设计

时间轴 ------> |--- 分钟1 ---|--- 分钟2 ---|--- 分钟3 ---| | | | | L1: [聚合中] L1: [聚合中] L1: [聚合中] | | | v v v L1结果 L1结果 L1结果 | | | +------+-------+------+------+ | | v v L2: [小时窗口聚合中...] | | v v L2下发 L2最终结果 (每分钟) (每小时) | v 写入存储 (小时粒度) 再经过24个小时窗口 | v 降采样为天粒度 写入存储 图3：L1→L2→存储的时序关系

五、为什么选择两级聚合？

5.1 一级聚合 vs 两级聚合

+-------------------------------------------------------------------+ | 一级聚合 vs 两级聚合 | +-------------------------------------------------------------------+ | | | 一级聚合（所有OAP节点直接写入存储，由存储层聚合） | | +----------------------------------------------------------------+ | | | 优点：架构简单，无需集群协调 | | | | 缺点： | | | | 1. 写入量大（每秒数万次写入） | | | | 2. 存储层压力巨大 | | | | 3. 无法实时查询分钟级指标（数据还没落入存储） | | | | 4. 不同存储实现的聚合能力不一致 | | | +----------------------------------------------------------------+ | | | | 两级聚合（OAP内存聚合 + 集群协调 + 批量写入） | | +----------------------------------------------------------------+ | | | 优点： | | | | 1. 写入量降低99%+（聚合后写入） | | | | 2. 存储层只负责持久化，不做计算 | | | | 3. OAP内存中可以直接查询实时指标 | | | | 4. 存储实现无关（H2/ES/MySQL一视同仁） | | | | 缺点： | | | | 1. OAP节点内存使用较大 | | | | 2. 集群协调复杂度增加 | | | | 3. 多节点崩溃时可能有数据丢失 | | | +----------------------------------------------------------------+ | | | +-------------------------------------------------------------------+ 图4：两级聚合的权衡分析

5.2 量化的性能收益

假设场景：100个服务，每秒10000个Segment请求 【一级聚合方案】 写入次数: 10000次/秒（原始Segment） 每日写入量: 864,000,000次 存储磁盘IO: ~86.4GB/天（假设每个Segment 100B） 【两级聚合方案】 L1聚合后写入: ~100次/秒（100个服务 × 聚合结果） 每日写入量: 8,640,000次 存储磁盘IO: ~2.5GB/天（按小时指标计） 性能提升： 写入量降低: 99% 磁盘IO降低: 97% CPU降低: ~90%（存储层不再需要聚合计算）

六、容错与数据一致性

两级聚合在极端情况下的行为：

6.1 OAP节点崩溃

正常情况： Node1 L1结果 -> L2聚合 -> 写入存储 ✓ Node2 L1结果 -> L2聚合 -> 写入存储 ✓ Node3 L1结果 -> L2聚合 -> 写入存储 ✓ Node2崩溃： Node1 L1结果 -> L2聚合 -> 写入存储 ✓ Node2 L1结果 -> 丢失（最多60秒数据） ✗ Node3 L1结果 -> L2聚合 -> 写入存储 ✓ 数据丢失范围： 最多丢失 Node2 上当前分钟窗口的L1聚合结果 大约相当于总数据的 1/N（N=节点数）

6.2 存储写入失败的退避

publicclassStorageWriteBuffer{privatefinalBlockingQueue<Metrics>pendingWrites;publicvoidwrite(Metricsmetrics)throwsStorageException{intretries=3;intbackoffMs=1000;for(inti=0;i<retries;i++){try{storageDAO.insert(metrics);return;// 成功}catch(StorageExceptione){logger.warn("写入存储失败(重试 {}/{}): {}",i+1,retries,e.getMessage());if(i<retries-1){Thread.sleep(backoffMs*(i+1));// 指数退避}}}// 三次重试都失败，放入pending队列pendingWrites.offer(metrics);logger.error("存储写入持久失败，数据已缓存到pending队列");}}

七、指标数据在两级聚合中的生命周期

Segment到达 | v [L1聚合 - 内存Map] 每收到一条数据就调用 combine(source, metrics) | | (每分钟窗口闭合) v +-----------+-----------+ | | v v L1结果 L1结果存入 发送给L2 本地小时缓存 | | | v | [本地小时聚合] | 合并多个L1分钟结果 | v [L2收集缓冲区] 收集所有节点的L1结果 | | (收集足够节点结果后) v [L2聚合] --> combine(L1_A, L1_B, L1_C...) | v [写入存储] |-- 分钟粒度结果 (可选) | v [小时聚合] 在L2层对分钟结果做二次聚合 | v [写入存储 - 小时粒度] | v [天级聚合] 在存储层通过小时数据降采样 | v [写入存储 - 天粒度] | v [TTL清理] 超过保留期的数据删除 图5：指标数据从生到死的完整生命周期

八、总结

两级聚合是SkyWalking在分布式可观测场景下的最优解：

这个架构的精妙之处在于：用内存换IO，用集群协调换数据量缩减，用局部容忍换全局性能。

下一篇，我们将深入OAP集群间的通信协议，看看L1结果是如何在节点间高效传输的。

上一篇【第44篇】自定义OAL指标实战：打造你的专属可观测仪表盘
下一篇【第46篇】OAP集群通信协议：节点间对话的暗语