在之前的《网易云信服务监控平台实践》一文中，我们环绕数据采集、数据处理、监控告警、数据运用 4 个环节，先容了网易云信服务监控平台的整体框架。本文是对网易云信在聚合指标打算逻辑上的进一步详述。
基于明细数据集进行实时聚合，生产一个聚合指标，业界常用的实现办法是 Spark Streaming、Flink SQL / Stream API。不论是何种办法，我们都须要通过写代码来指天命据来源、数据洗濯逻辑、聚合维度、聚合窗口大小、聚合算子等。如此繁杂的逻辑和代码，无论是开拓、测试，还是后续任务的掩护，都须要投入大量的人力/物力本钱。而我们程序员要做的便是化繁为简、实现大巧不工。
本文将阐述网易云信是如何基于 Flink 的 Stream API，实现一套通用的聚合指标打算框架。
如上图所示，是我们基于 Flink 自研的聚合指标完全加工链路，个中涉及到的模块包括：
source：定期加载聚合规则，并根据聚合规则按需创建 Kafka 的 Consumer，并持续消费数据。process：包括分组逻辑、窗口逻辑、聚合逻辑、环比打算逻辑等。从图中可以看到，我们在聚合阶段分成了两个，这样做的目的是什么？个中的好处是什么呢？做过分布式和并发打算的，都会碰着一个共同的仇敌：数据倾斜。在我们 PaaS 做事中头部客户会更加明显，以是倾斜非常严重，分成两个阶段进行聚合的奥妙下文中会详细解释。sink：是数据输出层，目前默认输出到 Kafka 和 InfluxDB，前者用于驱动后续打算（如告警关照等），后者用于数据展示以及查询做事等。reporter：全链路统计各个环节的运行状况，如输入/输出 QPS、打算耗时、消费堆积、迟到数据量等。下文将详细先容这几个模块的设计和实现思路。
3 source规则配置
为了便于聚合指标的生产和掩护，我们将指标打算过程中涉及到的关键参数进行了抽象提炼，供应了可视化配置页面，如下图所示。下文会结合详细场景先容各个参数的用场。
规则加载
在聚合任务运行过程中，我们会定期加载配置。如果检测到有新增的 Topic，我们会创建 kafka-consumer 线程，吸收上游实时数据流。同理，对付已经失落效的配置，我们会关闭消费线程，并清理干系的 reporter。
数据消费
对付数据源相同的聚合指标，我们共用一个 kafka-consumer，拉取到记录并解析后，对每个聚合指标分别调用 collect() 进行数据分发。如果指标的数据筛选规则（配置项⑤）非空，在数据分发前须要进行数据过滤，不知足条件的数据直接丢弃。
4 process整体打算流程
基于 Flink 的 Stream API 实现聚合打算的核心代码如下所示：
SingleOutputStreamOperator<MetricContext> aggResult = src .assignTimestampsAndWatermarks(new MetricWatermark()) .keyBy(new MetricKeyBy()) .window(new MetricTimeWindow()) .aggregate(new MetricAggFuction());MetricWatermark()：根据指定的韶光字段（配置项⑧）获取输入数据的 timestamp，并驱动打算流的 watermark 往前推进。MetricKeyBy()：指定聚合维度，类似于 MySQL 中 groupby，根据分组字段（配置项⑥），从数据中获取聚合维度的取值，拼接身分组 key。MetricTimeWindow()：配置项⑧中指定了聚合打算的窗口大小。如果配置了定时输出，我们就创建滑动窗口，否则就创建滚动窗口。MetricAggFuction()：实现配置项②指定的各种算子的打算，下文将详细先容各个算子的实现事理。
二次聚合
对付大数据量的聚合打算，数据倾斜是不得不考虑的问题，数据倾斜意味着规则中配置的分组字段（配置项⑥）指定的聚合 key 存在热点。我们的打算框架在设计之初就考虑了如何办理数据倾斜问题，便是将聚合过程拆分成2阶段：
第1阶段：将数据随机打散，进行预聚合。第2阶段：将第1阶段的预聚合结果作为输入，进行终极的聚合。详细实现：判断并发度参数 parallelism（配置项⑦） 是否大于1，如果 parallelism 大于1，天生一个 [0, parallelism) 之间的随机数作为 randomKey，在第1阶段聚合 keyBy() 中，将依据分组字段（配置项⑥）获取的 key 与 randomKey 拼接，天生终极的聚合 key，从而实现了数据随机打散。
聚合算子
作为一个平台型的产品，我们供应了如下常见的聚合算子。由于采取了二次聚合逻辑，各个算子在第1阶段和第2阶段采取了相应的打算策略。
算子
第1阶段聚合
第2阶段聚合
min/max/sum/count
直接对输入数据进行预聚合打算，输出预聚合结果
对第1阶段预聚合结果进行二次聚合打算，输出终极结果
first/last
对输入数据的 timestamp 进行比较，记录最小/最大的 timestamp 以及对应的 value 值，输出 <timestamp,value> 数据对
对 <timestamp,value> 数据对进行二次打算，输出终极的 first/last
avg
打算该分组的和值和记录数，输出 <sum,cnt> 数据对
对 <sum,cnt> 数据对分别求和，然后输出：总 sum / 总 cntcount
median/tp90/tp95
统计输入数据的分布，输出 NumericHistogram
对输入的 NumericHistogram 做 merge 操作，终极输出中位数/tp90/tp95
count-distinct
输出记录桶信息和位图的 RoaringArray
对 RoaringArray 进行 merge 操作，终极输出精确的去重计数结果
count-distinct(近似)
输出基数计数工具 HyperLoglog
对 HyperLoglog 进行 merge 操作，终极输出近似的去重计数结果
对付打算结果受全部数据影响的算子，如 count-distinct（去重计数），常规思路是利用 set 的去重特性，将所有统计数据放在一个 Set 中，终极在聚合函数的 getResult 中输出 Set 的 size。如果统计数据量非常大，这个 Set 工具就会非常大，对这个 Set 的 I/O 操作所花费的韶光将不能接管。
对付类 MapReduce 的大数据打算框架，性能的瓶颈每每涌如今 shuffle 阶段大工具的 I/O 上，由于数据须要序列化 / 传输 / 反序列化，Flink 也不例外。类似的算子还有 median 和 tp95。
为此，须要对这些算子做专门的优化，优化的思路便是只管即便减少打算过程中利用的数据工具的大小，个中：
median/tp90/tp95：参考了 hive percentile_approx 的近似算法，该算法通过 NumericHistogram（一种非等距直方图）记录数据分布，然后通过插值的办法得到相应的 tp 值（median 是 tp50）。count-distinct：采取 RoaringBitmap 算法，通过压缩位图的办法标记输入样本，终极得到精确的去重计数结果。count-distinct(近似) ：采取 HyperLoglog 算法，通过基数计数的办法，得到近似的去重计数结果。该算法适用于大数据集的去重计数。后处理
后处理模块，是对第2阶段聚合打算输出数据进行再加工，紧张有2个功能：
复合指标打算：对原始统计指标进行组合打算，得到新的组合指标。例如，要统计登录成功率，我们可以先分别统计出分母（登录次数）和分子（登录成功的次数），然后将分子除以分母，从而得到一个新的组合指标。配置项③便是用来配置组合指标的打算规则。相对指标打算：告警规则中常常要判断某个指标的相对变革情形（同比/环比）。我们利用 Flink 的state，能够方便的打算出同比/环比指标，配置项④便是用来配置相对指标规则。非常数据的处理
这里所说的非常数据，分为两类：迟到的数据和提前到的数据。
迟到数据： 对付严重迟到的数据（大于聚合窗口的 allowedLateness），通过 sideOutputLateData 进行网络，并通过 reporter 统计上报，从而能够在监控页面进行可视化监控。 对付轻微迟到的数据（小于聚合窗口的 allowedLateness），会触发窗口的重打算。如果每来一条迟到数据就触发一次第 1 阶段窗口的重打算，重打算结果传导到第 2 阶段聚合打算，就会导致部分数据的重复统计。为理解决重复统计的问题，我们在第 1 阶段聚合 Trigger 中进行了分外处理：窗口触发采取 FIRE_AND_PURGE（打算并清理），及时清理已经参与过打算的数据。提前到的数据：这部分数据每每是数据上报真个时钟不准导致。在打算这些数据的 timestamp 时要人为干预，避免影响全体打算流的 watermark。5 sink聚合打算得到的指标，默认输出到 Kafka 和时序数据库 InfluxDB。
kafka-sink：将指标标识（配置项①）作为 Kafka 的topic，将聚合结果发送出去，下贱吸收到该数据流后可以进一步处理加工，如告警事宜的生产等。InfluxDB-sink：将指标标识（配置项①）作为时序数据库的表名，将聚合结果持久化下来，用于 API 的数据查询、以及可视化报表展示等。6 reporter为了实时监控各个数据源和聚合指标的运行情形，我们通过 InfluxDB+Grafana 组合，实现了聚合打算全链路监控：如各环节的输入/输出 QPS、打算耗时、消费堆积、迟到数据量等。
7 结语
目前，通过该通用聚合框架，承载了网易云信 100+ 个不同维度的指标打算，带来的收益也是比较可不雅观的：
提效：采取了页面配置化办法实现聚合指标的生产，开拓周期从天级缩短到分钟级。没有数据开拓履历的同学也能够自己动手完成指标的配置。掩护大略，资源利用率高：100+ 个指标只需掩护 1 个 flink-job，资源花费也从 300+ 个 CU 减少到 40CU。运行过程透明：借助于全链路监控，哪个打算环节有瓶颈，哪个数据源有问题，一览无余。作者先容圣少友，网易云信数据平台资深开拓工程师，从事数据平台干系事情，卖力做事监控平台、数据运用平台、质量做事平台的设计开拓事情。
