基于Antlr在Apache Flink中實現監控規則DSL化的探索實踐
原創【51CTO.com原創稿件】1 引言
目前業界已經有很多杰出的監控系統如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 Time
- FOR 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 系統拆分、商品詳情頁接入層優化、云信客服系統重構等重大技術攻關項目。現致力于打造蘇寧易購新一代核心購物流程系統，希望將購物體驗做到更好。
胡正林，蘇寧科技集團消費者平臺高級架構師，十余年軟件開發經驗，熟悉大型分布式高并發系統架構和開發，目前主要負責易購各系統架構優化與大促保障工作。
【51CTO原創稿件，合作站點轉載請注明原文作者和出處為51CTO.com】
