基于Antlr在Apache Flink中实现监控规则DSL化的探索实践

基于Antlr在Apache Flink中实现监控规则DSL化的探索实践

目前业界已经有很多杰出的监控系统如Nagios、Zabbix、Prometheus，但都是面向运维人员，在应对复杂多变的业务指标监控时显得不够灵活。

为此苏宁数据云在流式计算框架Apache Flink之上设计了一组包含ETL、指标计算、告警触发、告警通知模块的业务监控引擎。

其基本规则是告警SQL DSL语言，定义完备易用的场景规则语法和算子并支持动态更新调整，以便于业务方接入。下面对其基本原理和实现进行介绍，供各位同行参考指正。

2 DSL规则设计

初期调研了各个业务方的需求，整理归纳出核心需求点主要是基于维度和时间的指标绝对值、同环比或方差在超出阈值范围则触发告警，且业务方接入的是原始明细数据，故在设计规则中需要包含数据清洗过滤、分组、聚合计算、时间窗口设置特性。

而SQL语言作为开发人员最熟悉的数据处理语言，选择其作为原型，可以省去理解和沟通的成本。

语法规则如下：

SELECT {metrics} FROM {metricFilters} WHERE {alertConditions} GROUP BY {groupByExpr} ORDER BY {orderByExpr} FOR LAST {number} MINUTE

SELECT子句指标计算FROM子句数据过滤表达式WHERE子句告警状态判断表达式GROUP BY子句数据分组ORDER BY子句排序规则FOR LAST子句计算窗口时间

目前聚合函数支持的均值(avg)、记数(count)、总数(sum)、去重记数(distinct_count)等指标计算，并支持多个指标四则混合运算、位运算、比较运算、逻辑运算。

以下SQL规则表达的业务含义是根据错误类型和城市信息进行分组，5分钟内错误数量大于100且与前5分钟环比大于20%则触发告警：

SELECT COUNT(errorType, 5) AS currentTypeNum, COUNT(errorType, 10, 5) AS previewTypeNum, (currentTypeNum - previewTypeNum) / previewTypeNumAS circularPercent, errorType, city FROM mobileAppIdIN ("Suning_PCbrowser","Suning_WAP")   WHERE currentTypeNum>100 AND circularPercent>0.2   GROUP BY errorType, city ORDER BY EVENT_TIME  FOR LAST10 MINUTE

3 实现设计

3.1 SQL解析

基于Antlr定义词法和语法，使用Maven Antlr插件生成AST解析器，Visitor模式遍历语法树实现各个部分处理器，不熟悉的同学可以参考Antlr官方手册，此处不再赘述。

解析SQL语句过程如下：

提取所有计算中用的到字段和过滤条件(FROM语句)，作为预处理器用于ETL过程WHERE子句抽象为布尔表达式，系统状态判断条件GROUP BY子句抽象为从一个数据对象中提取分组Key的处理器ORDER BY从句中提取时间窗口字段即Apache Flink中Event Time或Process TimeFOR LAST子句提取出时间窗口大小SELECT语句是一组支持AS别名的算子，执行后的结果为键值对，即业务指标

需要注意的是考虑到ETL过程和计算过程都是独立模块，若计算过程通过字段名引用字段值，则中间对象必然是键值对方式的Map结构，在传输中序列化和反序列化必然会对性能有一定的影响。

为此在遍历抽象语法树过程中须构建一个解析上下文，把字段名引用改为数组下标，ETL过程产生的中间数据对象为数组结构，计算时访问字段值的时间复杂度为O(1)。

例如原始数据为：

{"errorType":"E005","mobileAppId":"Suning_WAP","city":"025","network":"CMCC"} 解析出子字段数组[“errorType”,”city”]发送ETL模块，处理后数据对象Row为[“E005”,”025”]，算子COUNT(“errorType”)实际运行中为COUNT(ROW[0])

3.2 整体架构

整个流程中ETL和告警计算模块都是运行在Apache Flink中，借用Flink实时计算和状态持久化能力。

每个业务接入方的数据格式不尽相同，把相关数据解析、清洗、过滤、丰富等功能单独抽离成ETL模块，可以根据接入方业务需求单独定制部署，其中SQL中FROM阶段提前到ETL模块，提取过滤掉不需要的数据，降低数据传输量。

3.3 SQL引擎在Flink中的运行机制

ETL模块

告警规则SourceStream转成广播流BroadcastStream并connect到数据解析Stream，因并行度不一致且未分组须使用BroadcastProcessFunction处理规则的变化通知，把SQL规则解析成字段提取和过滤处理器，保存在BroadcastState中达到数据流和广播流共享状态的目的。

告警计算模块

从Kafka中消费ETL模块提取的字段和规则ID，根据告警规则Group子句提取分组信息;结合规则流解析Select子句表达式计算所有指标;再次结合规则流解析Where 子句告警触发条件和状态机相关参数，判断系统当前健康状态，触发健康状态机变迁;把系统状态变迁和当前指标值作为告警事件写入Kakfa，由告警通知模块根据通知规则配置处理相关发送逻辑。

3.4 聚合计算

Apache Flink是流式实时处理框架，而聚合计算类似于批处理，需要计算窗口内的所有数据;相对应告警SQL可能会产生大量分组，每个分组其实都是独立的告警规则，若是缓存一个事件窗口的数据进行计算，会对Flink的状态维护产生巨大压力，所以对于聚合计算采用的分钟级分桶计算和累加器聚合结果设计思路。

分桶

根据时间窗口长度创建环形队列，每分钟一个桶，按照时间计算当前元素桶的位置，计算当前桶的值。

累加器

类似于Spark和Flink中累加器Accumulator实现，保存分钟级计算中间结果、合并累加器、获取最终值，因此计算过程中不必保留原始数据，只需要在Flink中保存累加器的状态即可。

/**      * Creates a new accumulator, starting a new aggregate.      */      ADD createAccumulator();       /**      * Adds the given input value to the given accumulator, returning the      * new accumulator value.      */ ADD add(ELE value, ADD accumulator);  /**      * Gets the result of the aggregation from the accumulator.      */ OUT getResult(ADD accumulator);  /**      * Merges two accumulators, returning an accumulator with the merged state.      */ ADD merge(ADD a, ADD b); }

其中去重计数distinct_count聚合计算在告警场景允许精度损失故采用支持分桶累加的HyperLogLog算法，降低对内存的要求。

3.5 告警状态机

是否产生告警事件并不是依据计算出的指标，而是根据Where从句判断指标是否超出阈值返回True或False确定当前系统健康状态，比对当前状态产生的健康状态变迁事件作为触发告警事件。

目前设计系统的状态有三种，即正常(Normal)、警告(Warning)、严重(Critical)，而后两种就是分别对应两个SQL DSL中Where从句表达式，状态变迁图如下：

业务系统是否产生告警通知也可以根据这几种事件进行配置，如果连续5分钟异常才产生告警，则可以只关注CONTINUE相关事件。当然告警通知系统也有通知合并和告警风暴抑制相关功能。

4 问题和展望

目前此引擎已经用于苏宁易购登录、商品详情页、购物车和支付多个业务线的用户体验监控，帮助产品和业务运营人员可以快速发现和定位问题。为了提升处理能力和降低接入难度，以下两个方面还需要进行优化：

改进Apache Flink状态管理 现在Flink集群中状态管理使用的FsStateBackend机制,状态保存在HDFS文件系统，在高基数维度或多维组合产生海量分组的业务场景下可能会有OOM和性能风险，下一步准备切换为支持增量Checkpoint的RocksDB状态存储方案。异常检测智能化 现有告警规则如何判断系统状态是否异常还是依据静态阈值，这种方式要求业务方对系统指标有精确的了解，且随着业务的变化要不断调整，不符合业界的发展趋势。下一阶段计划针对新业务无历史数据的场景采用曲线波动检测3sigma和中位数绝对偏差MAD(Median absolute deviation)算法，而比较稳定的业务线引入时间序列异常检测机器学习。

苏宁数据云服务产品

苏宁数据云已有大数据开发套件(提供全方位的大数据开发服务)、人工智能服务(实现智能的人机互动做出更好决策)、数据分析及展示服务(提供海量数据处理与分析方法)、平台基础服务(提供大数据平台基础服务)，实时告警数据分析平台也在规划建设中。实时计算中实现监控规则DSL是内部项目的一次技术探索，若能经过实际业务的考验，也会考虑通过数据云对外提供服务。

作者简介

黄小虎，苏宁科技集团消费者平台购物流程架构负责人，全面负责苏宁易购商品详情页、购物车、大聚会等核心系统的优化及大促保障工作。对电商交易流程和业务有较深入的思考和研究，专注于高并发大型电商网站的架构设计、高可用的系统设计。曾主导和参与了 Commerce 系统拆分、商品详情页接入层优化、云信客服系统重构等重大技术攻关项目。现致力于打造苏宁易购新一代核心购物流程系统，希望将购物体验做到更好。

胡正林，苏宁科技集团消费者平台高级架构师，十余年软件开发经验，熟悉大型分布式高并发系统架构和开发，目前主要负责易购各系统架构优化与大促保障工作。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。