基于Flink+QLExpress處理海量數(shù)據(jù)的實(shí)時(shí)監(jiān)控告警系統(tǒng)

陳 麗 楊大龍 高中纖 張 偉

中國聯(lián)合網(wǎng)絡(luò)通信有限公司濟(jì)南軟件研究院 濟(jì)南 250100

引言

隨著物聯(lián)網(wǎng)、移動(dòng)互聯(lián)網(wǎng)的發(fā)展及信息技術(shù)向各行業(yè)的深入滲透，隨之而來的是海量數(shù)據(jù)的存儲(chǔ)、計(jì)算和分析等問題，而海量數(shù)據(jù)的處理給運(yùn)維監(jiān)控系統(tǒng)的核心告警系統(tǒng)帶來了極大的挑戰(zhàn)。目前的海量數(shù)據(jù)監(jiān)控告警系統(tǒng)通常采用Kafka+Flink模式，此模式不夠靈活。前人對(duì)海量數(shù)據(jù)流處理的研究多集中在Flink技術(shù)上，在實(shí)際生產(chǎn)過程中應(yīng)用Flink來處理海量日志，提升數(shù)據(jù)處理效率及能力，提高告警的及時(shí)性。但是如上傳統(tǒng)模式的監(jiān)控告警系統(tǒng)無法動(dòng)態(tài)更新告警口徑、閾值、格式及信息推送方式等，往往需要通過更新版本代碼的方式實(shí)現(xiàn)告警口徑等的增加或修改。

現(xiàn)有的大數(shù)據(jù)監(jiān)控告警系統(tǒng)的實(shí)現(xiàn)方式可以概括為如下四類[1-3]。一是直接基于Flink實(shí)現(xiàn)監(jiān)控告警，對(duì)于數(shù)據(jù)的獲取方式采用傳統(tǒng)方式，F(xiàn)link作為實(shí)時(shí)流計(jì)算引擎對(duì)數(shù)據(jù)進(jìn)行處理分析得到中間結(jié)果數(shù)據(jù)，然后與閾值進(jìn)行比較，符合閾值條件的發(fā)送告警信息；二是Kafka作為消息中間件，其不同Topic的數(shù)據(jù)作為Flink的數(shù)據(jù)源，F(xiàn)link負(fù)責(zé)依據(jù)預(yù)設(shè)的指標(biāo)進(jìn)行數(shù)據(jù)處理及過濾，并將結(jié)果寫入Redis[4]，Promethues從Redis拉取數(shù)據(jù)，Grafana根據(jù)數(shù)據(jù)進(jìn)行圖表展示；三是利用基于QLExpress規(guī)則引擎的監(jiān)控告警系統(tǒng)，可以實(shí)現(xiàn)靈活預(yù)警、配置動(dòng)態(tài)發(fā)布與更新，但是無法承載海量數(shù)據(jù)的實(shí)時(shí)分析及處理；四是基于Flink的實(shí)時(shí)計(jì)算平臺(tái)的設(shè)計(jì)與實(shí)現(xiàn)，使用Flink提供的SQL語法功能實(shí)現(xiàn)配置的動(dòng)態(tài)更新，搭建了一個(gè)可視化的實(shí)時(shí)計(jì)算平臺(tái)，但未將其合理地運(yùn)用到監(jiān)控告警系統(tǒng)中。因此，前人對(duì)告警系統(tǒng)的研究多集中在Flink或者QLExpress上，而對(duì)Flink+QLExpress技術(shù)模式處理海量數(shù)據(jù)的研究較少，還有待深入。

本文中通過引入Flink SQL和QLExpress技術(shù)優(yōu)化現(xiàn)有大數(shù)據(jù)監(jiān)控告警系統(tǒng)架構(gòu)，實(shí)現(xiàn)告警口徑等的靈活配置及動(dòng)態(tài)更新，解決告警無法動(dòng)態(tài)配置、海量數(shù)據(jù)處理效率慢、告警實(shí)時(shí)性不足等問題?；舅悸肥抢肅anal采集原始業(yè)務(wù)數(shù)據(jù)，通過自定義聚合程序聚合離散的業(yè)務(wù)數(shù)據(jù)推送至Kafka，Kafka的數(shù)據(jù)作為Flink的數(shù)據(jù)源，然后Flink獲取數(shù)據(jù)后利用Flink SQL技術(shù)實(shí)現(xiàn)數(shù)據(jù)清洗及過濾得到初級(jí)告警數(shù)據(jù)，然后通過QLExpress技術(shù)實(shí)現(xiàn)告警數(shù)據(jù)的二次過濾及告警信息的格式化、推送等，最終完成海量數(shù)據(jù)的監(jiān)控告警。同時(shí)，該告警系統(tǒng)提供管理控制臺(tái)，方便對(duì)各類腳本的管理及歷史告警信息的查詢與統(tǒng)計(jì)分析。

1 核心技術(shù)

本文涉及的核心技術(shù)共四項(xiàng)，其中核心設(shè)計(jì)思路中的關(guān)鍵核心技術(shù)包括Canal數(shù)據(jù)采集工具、Flink技術(shù)、Flink SQL技術(shù)及QLExpress技術(shù)。為實(shí)現(xiàn)告警口徑、閾值、信息的格式、推送方式的動(dòng)態(tài)更新，海量數(shù)據(jù)的處理與分析，業(yè)務(wù)邏輯與告警處理程序的解耦，F(xiàn)link技術(shù)、Flink SQL技術(shù)及QLExpress技術(shù)起到至關(guān)重要的作用，下面將分別進(jìn)行詳細(xì)介紹。

1.1 Canal技術(shù)

Canal是阿里開源的基于MySQL數(shù)據(jù)庫增量日志解析，提供增量數(shù)據(jù)訂閱和消費(fèi)，類似MySQL的一個(gè)Slave，實(shí)現(xiàn)數(shù)據(jù)的同步與備份，進(jìn)而達(dá)到數(shù)據(jù)采集的目的。其工作原理圖如圖1所示。

圖1 Canal工作原理圖

1)Canal模擬MySQL Slave的交互協(xié)議，將自己裝扮為一個(gè)MySQL Slave，并將dump協(xié)議轉(zhuǎn)發(fā)至MySQL Master。

2)MySQL Master接收到Slave的轉(zhuǎn)出請(qǐng)求后，開始將二進(jìn)制日志推送給Slave，即Canal。

3)Canal解析二進(jìn)制日志為自己支持的數(shù)據(jù)類型，后續(xù)可通過加工處理后作為不同的對(duì)象的源數(shù)據(jù)，如MySQL、Kafka、Elasticsearch、Hbase等。

Canal適用的場景如下：1)數(shù)據(jù)同步，如實(shí)現(xiàn)在線、離線數(shù)據(jù)庫之間的數(shù)據(jù)同步。2)數(shù)據(jù)消費(fèi)，如對(duì)所關(guān)注的數(shù)據(jù)庫表產(chǎn)生的變化，實(shí)現(xiàn)數(shù)據(jù)增量搜索。3)數(shù)據(jù)脫敏，如將線上動(dòng)態(tài)數(shù)據(jù)導(dǎo)入到其他存儲(chǔ)結(jié)構(gòu)，實(shí)現(xiàn)數(shù)據(jù)的脫敏。

Canal具有HA機(jī)制，是Canal實(shí)現(xiàn)高可用的關(guān)鍵技術(shù)，本文使用了Canal的這一優(yōu)良特性，實(shí)現(xiàn)Canal與數(shù)據(jù)源之間1對(duì)多的配置關(guān)系。

1.2 Kafka技術(shù)

Apache Kafka最初是由LinkedIn研發(fā)的一個(gè)分布式的發(fā)布—訂閱消息系統(tǒng)[5-6]，為處理實(shí)時(shí)數(shù)據(jù)提供一個(gè)統(tǒng)一、高通量、低等待的平臺(tái)。Kafka是一個(gè)分布式系統(tǒng)，由通過高性能TCP網(wǎng)絡(luò)協(xié)議進(jìn)行通信的服務(wù)器和客戶端組成，可部署在內(nèi)部以及云環(huán)境中的裸機(jī)硬件、虛擬機(jī)和容器上。Kafka集群可以在一個(gè)指定的時(shí)間內(nèi)保存所有發(fā)布上來的消息，當(dāng)超過指定的時(shí)間后消息就會(huì)被系統(tǒng)丟棄以釋放空間。Kafka的性能不會(huì)受數(shù)據(jù)量大小的影響，因此可以保存大量的數(shù)據(jù)。

1.3 Flink及Flink SQL技術(shù)

Flink是一個(gè)針對(duì)流數(shù)據(jù)和批數(shù)據(jù)的分布式處理引擎[7-10]，所要處理的場景主要是流數(shù)據(jù)，會(huì)把所有任務(wù)當(dāng)做流來處理。因此，F(xiàn)link可以提供數(shù)據(jù)處理的通用方案，使得由相同或相關(guān)的組件和API搭建框架的處理需求得以簡化。同時(shí)，F(xiàn)link可以支持本地快速迭代，可以定制化內(nèi)存管理。所以，F(xiàn)link的特點(diǎn)是處理大數(shù)據(jù)速度快，能滿足大數(shù)據(jù)背景下應(yīng)用實(shí)時(shí)計(jì)算平臺(tái)的需求。

Flink是一個(gè)真正的流式處理，其可以按照元組逐個(gè)分發(fā)數(shù)據(jù)，做到對(duì)數(shù)據(jù)的單條處理，而不需要再流處理匯總執(zhí)行計(jì)算。其支持大規(guī)模計(jì)算能力，能夠在數(shù)千個(gè)節(jié)點(diǎn)上并發(fā)運(yùn)行，具有強(qiáng)大的吞吐量和延遲特性，提供了數(shù)據(jù)分布、數(shù)據(jù)通信以及容錯(cuò)機(jī)制等功能。同時(shí)，F(xiàn)link提供多種靈活的窗口函數(shù)及更高抽象層的API，使開發(fā)者可以更便捷地編寫分布式任務(wù)。

Flink SQL是面向用戶的API層[11]，是降低用戶使用實(shí)時(shí)計(jì)算門檻而設(shè)計(jì)的一套符合標(biāo)準(zhǔn)SQL語義的開發(fā)語言，開發(fā)者可以直接在代碼中書寫SQL實(shí)現(xiàn)一些查詢操作。Flink的SQL支持基本查詢語法，常見的算子包括SELECT、WHERE、DISTINCT、GROUP BY、UNION、UNION ALL、JOIN等。同時(shí)，F(xiàn)link SQL提供大量的內(nèi)置函數(shù)，常見的內(nèi)置函數(shù)包括比較函數(shù)、邏輯函數(shù)、算術(shù)函數(shù)、字符串處理函數(shù)、時(shí)間函數(shù)，也支持使用者自定義函數(shù)實(shí)現(xiàn)其他計(jì)算。Flink實(shí)現(xiàn)了SQL標(biāo)準(zhǔn)的Apache Calcite(Apache開源 SQL解析工具)，語法易讀易書寫，開發(fā)者只需關(guān)注業(yè)務(wù)邏輯。而且Flink內(nèi)置很多優(yōu)化規(guī)則，簡化了研發(fā)的復(fù)雜度，通過SQL能在高層應(yīng)用上實(shí)現(xiàn)真正的批流一體。

1.4 QLExpress技術(shù)

QLExpress是一個(gè)開源的類Java語法的規(guī)則引擎[12-13]，作為一個(gè)嵌入式規(guī)則引擎在應(yīng)用程序中使用。該規(guī)則引擎是一種動(dòng)態(tài)腳本解析引擎，用Java語言來編寫腳本，支持自定義運(yùn)算符、自定義函數(shù)、宏定義、拓展操作等，可以簡化業(yè)務(wù)規(guī)則，從而使得規(guī)則的定義變的靈活。其實(shí)現(xiàn)原理主要包括定義規(guī)則、解析規(guī)則、執(zhí)行規(guī)則三個(gè)步驟，編譯過程是一個(gè)實(shí)現(xiàn)了完整的詞法分析、語法分析、語義分析、編譯指令、指令執(zhí)行的過程，類似Java語言中Class文件的編譯過程。

QLExpress規(guī)則引擎被應(yīng)用在各類場景中，其特性如下。1)線程安全：在引擎運(yùn)算過程中產(chǎn)生的所有臨時(shí)變量都是ThreadLocal類型；2)執(zhí)行效率高：將耗時(shí)長的腳本編譯過程緩存在本地，運(yùn)行時(shí)臨時(shí)變量的創(chuàng)建使用緩沖池技術(shù)實(shí)現(xiàn)，其性能與Groovy性能相當(dāng)；3)弱類型腳本語言：同JavaScript及Groovy的語法類似，其運(yùn)行速率慢于強(qiáng)類型腳本語言，但其大大增強(qiáng)了業(yè)務(wù)實(shí)現(xiàn)及配置的靈活度；4)安全控制：通過相關(guān)運(yùn)行參數(shù)的配置防止死循環(huán)及惡意調(diào)用等異常情況的發(fā)生。

2 系統(tǒng)設(shè)計(jì)

目前，大數(shù)據(jù)監(jiān)控告警系統(tǒng)采用的方案通常為下述三種方案：方案一是采用Filebeat+Kafka+Flink+告警的架構(gòu)，無法實(shí)現(xiàn)告警口徑的動(dòng)態(tài)更新及與業(yè)務(wù)代碼邏輯的解耦；方案二是采用Filebeat+Kafka+Storm[14]+QLExpress+告警的架構(gòu)，Storm流式處理采用ACK機(jī)制，對(duì)每個(gè)消息進(jìn)行全鏈路跟蹤，失敗或超時(shí)進(jìn)行重發(fā)，每次需要一個(gè)ACK機(jī)制，無法實(shí)現(xiàn)真正的流處理，相比Flink存在一定的劣勢；方案三是Filebeat+Kafka+Flink+QLExpress+告警的架構(gòu)，該架構(gòu)幾近完美，其中Filebeat是一種容器日志采集利器，讀取的對(duì)象是日志文件，對(duì)于MySQL的二進(jìn)制文件無法讀取，不適用于對(duì)源數(shù)據(jù)存儲(chǔ)于MySQL的系統(tǒng)。

本文提出的基于Canal+Kafka+Flink+Flink SQL+QLExpress+Redis+告警的處理海量數(shù)據(jù)的監(jiān)控告警系統(tǒng)由數(shù)據(jù)采集模塊、數(shù)據(jù)聚合模塊、數(shù)據(jù)處理模塊、告警推送模塊、數(shù)據(jù)存儲(chǔ)模塊、管理控制臺(tái)模塊六部分組成。數(shù)據(jù)采集模塊是通過日志采集工具將日志庫中存儲(chǔ)的業(yè)務(wù)告警原始數(shù)據(jù)采集至聚合程序中；數(shù)據(jù)聚合模塊是獲取來自數(shù)據(jù)采集模塊已采集的業(yè)務(wù)數(shù)據(jù)，通過業(yè)務(wù)邏輯代碼程序?qū)I(yè)務(wù)數(shù)據(jù)依據(jù)具體需求進(jìn)行拼接封裝；數(shù)據(jù)處理模塊是該監(jiān)控告警系統(tǒng)的核心，先進(jìn)行數(shù)據(jù)獲取，再通過腳本進(jìn)行數(shù)據(jù)清洗及數(shù)據(jù)過濾，依據(jù)閾值條件得到初級(jí)告警數(shù)據(jù)；告警推送模塊是通過腳本將初級(jí)告警數(shù)據(jù)依據(jù)不同閾值進(jìn)行二次數(shù)據(jù)過濾，依據(jù)告警信息模板生成可讀的告警信息，然后通過調(diào)用不同應(yīng)用系統(tǒng)的API將告警信息推送至不同的應(yīng)用系統(tǒng)(如釘釘、微信、電話等)；數(shù)據(jù)存儲(chǔ)模塊是將已推送的告警信息進(jìn)行存儲(chǔ)，用于后續(xù)告警數(shù)據(jù)統(tǒng)計(jì)與分析；管理控制臺(tái)模塊負(fù)責(zé)管理此監(jiān)控告警系統(tǒng)中數(shù)據(jù)處理模塊及告警推送模塊中使用的腳本，以此完成整個(gè)告警流程。本監(jiān)控告警系統(tǒng)的技術(shù)架構(gòu)如圖2所示。

圖2 基于Fink+QLExpress處理海量數(shù)據(jù)的告警系統(tǒng)技術(shù)架構(gòu)圖

2.1 數(shù)據(jù)采集

數(shù)據(jù)采集模塊包括數(shù)據(jù)源及采集工具。數(shù)據(jù)源一般是MySQL數(shù)據(jù)庫，存儲(chǔ)的是通過平臺(tái)產(chǎn)生的實(shí)時(shí)交易數(shù)據(jù)，如能力的基本信息(能力名稱、版本號(hào)、交易接收時(shí)間、交易處理結(jié)束時(shí)間等)、交易響應(yīng)碼(平臺(tái)返回碼、業(yè)務(wù)返回碼等)、主子服務(wù)的關(guān)聯(lián)關(guān)系字段等。Canal采用高可用模式，即一個(gè)Canal實(shí)例采集多個(gè)日志庫數(shù)據(jù)，實(shí)現(xiàn)源數(shù)據(jù)的采集。

2.2 數(shù)據(jù)聚合

數(shù)據(jù)聚合模塊包括聚合程序及消息中間件。聚合程序用于將Canal采集得到的數(shù)據(jù)進(jìn)行主子服務(wù)的聚合、所需字段的獲取及整理，然后將聚合結(jié)果推送至Kafka的某個(gè)Topic。

2.3 數(shù)據(jù)處理

數(shù)據(jù)處理模塊包括數(shù)據(jù)獲取、數(shù)據(jù)清洗及過濾。首先進(jìn)行數(shù)據(jù)獲取，F(xiàn)link將Kafka作為數(shù)據(jù)源，讀取Topic的數(shù)據(jù)，將Kafka的Json格式的數(shù)據(jù)注冊(cè)為一個(gè)實(shí)體并存入自定義表中；然后進(jìn)行數(shù)據(jù)清洗及過濾，從Redis中讀取已存入的Flink SQL腳本，通過該SQL腳本進(jìn)行一次過濾，得到1分鐘粒度的告警數(shù)據(jù)，即初級(jí)告警數(shù)據(jù)，然后從Redis中讀取已存入的QLExpress腳本，通過QLExpress腳本對(duì)初級(jí)告警數(shù)據(jù)進(jìn)行二次過濾。如上兩個(gè)腳本是程序首次啟動(dòng)后會(huì)從數(shù)據(jù)存儲(chǔ)模塊獲取，同時(shí)寫入Redis中，后續(xù)只需從Redis中獲取即可。

2.4 告警推送

告警推送模塊包括告警信息格式化及告警推送。采用QLExpress技術(shù)進(jìn)行告警信息格式化及實(shí)現(xiàn)告警推送，該技術(shù)支持Java語法，簡單易學(xué)，告警配置人員能夠快速配置告警腳本。QLExpress腳本中先對(duì)告警數(shù)據(jù)進(jìn)行格式轉(zhuǎn)換，轉(zhuǎn)為用戶能夠清晰可見的文字描述，然后調(diào)用微信API、釘釘API進(jìn)行數(shù)據(jù)推送，運(yùn)維人員只需關(guān)注對(duì)應(yīng)的釘釘群或者微信群獲取告警信息，如圖3所示。同時(shí)，該告警系統(tǒng)支持告警外呼功能，即通過電話方式將告警信息通知到當(dāng)日值班人員，同時(shí)外呼告警歷史可查詢。

圖3 告警信息截圖

2.5 數(shù)據(jù)存儲(chǔ)

數(shù)據(jù)存儲(chǔ)模塊中存儲(chǔ)了兩類信息，一是已推送的告警信息，用于歷史告警數(shù)據(jù)的查詢，是管理控制臺(tái)模塊中告警日志管理頁面所展示數(shù)據(jù)的來源；二是FLink SQL腳本及QLExpress腳本。

2.6 管理控制臺(tái)

管理控制臺(tái)模塊包含大數(shù)據(jù)告警SQL管理(腳本管理)及告警日志管理，前者管理數(shù)據(jù)處理模塊中使用FLink SQL數(shù)據(jù)清洗、過濾腳本及QLExpress告警推送腳本，后者用于告警歷史數(shù)據(jù)的查看。

3 應(yīng)用效果

基于此新架構(gòu)的監(jiān)控告警系統(tǒng)已應(yīng)用在中國聯(lián)通天擎項(xiàng)目中，實(shí)現(xiàn)對(duì)海量能力交易數(shù)據(jù)的實(shí)時(shí)分析及處理，完成能力異常調(diào)用的實(shí)時(shí)監(jiān)控告警，極大提升了告警及時(shí)性，增強(qiáng)了告警配置的靈活性，目前已取得了顯著的成效，主要體現(xiàn)在以下幾個(gè)方面：1)每分鐘處理的數(shù)量達(dá)到261萬+條；2)可實(shí)現(xiàn)秒級(jí)告警；3)告警準(zhǔn)確率達(dá)到100%；4)處理的應(yīng)用系統(tǒng)量達(dá)2000+個(gè)；5)節(jié)省運(yùn)維成本達(dá)到120人/年，由8人年降低至2人年；6)告警配置速率提升24倍，由2天降低至2小時(shí)。

4 結(jié)語

本文提出的基于Flink+QLExpress的處理海量數(shù)據(jù)的告警系統(tǒng)，創(chuàng)造性地將Flink SQL及QLExpress、Redis技術(shù)同時(shí)引入監(jiān)控告警系統(tǒng)，解決了告警口徑的變更與代碼實(shí)現(xiàn)邏輯是緊耦合、告警無法靈活配置、配置無法動(dòng)態(tài)更新等問題。Flink技術(shù)實(shí)現(xiàn)億級(jí)數(shù)據(jù)的實(shí)時(shí)處理，F(xiàn)link SQL實(shí)現(xiàn)了告警口徑的變更與代碼實(shí)現(xiàn)邏輯的解耦，QLExpress的引入實(shí)現(xiàn)了告警信息的樣式及告警推送方式的靈活配置，兩者結(jié)合Redis技術(shù)實(shí)現(xiàn)配置的動(dòng)態(tài)更新。兩類腳本的書寫方式簡單、易讀，節(jié)省了研發(fā)成本，提升了研發(fā)效率。同時(shí)，本監(jiān)控告警系統(tǒng)的多租戶模式將不同系統(tǒng)的告警信息進(jìn)行數(shù)據(jù)隔離，實(shí)現(xiàn)數(shù)據(jù)的安全性及隔離性，滿足不同告警需求。

實(shí)踐表明，本文提出的基于全新架構(gòu)的監(jiān)控告警系統(tǒng)實(shí)現(xiàn)了對(duì)億級(jí)業(yè)務(wù)數(shù)據(jù)的實(shí)時(shí)采集及分析處理，實(shí)現(xiàn)告警口徑、告警閾值、告警格式及推送方式的動(dòng)態(tài)發(fā)布與更新，并通過可視化腳本實(shí)現(xiàn)數(shù)據(jù)過濾及告警信息推送功能的編寫，提升了告警研發(fā)速率。此監(jiān)控告警系統(tǒng)是天擎能力開放平臺(tái)正在使用的智能監(jiān)控告警系統(tǒng)，增強(qiáng)了平臺(tái)的預(yù)警能力，提升了告警及時(shí)性，保證了天擎能力開放平臺(tái)的穩(wěn)定高效運(yùn)行，應(yīng)用效果顯著。