基于大數(shù)據(jù)技術(shù)的氣象業(yè)務(wù)監(jiān)視數(shù)據(jù)采集處理

曾 樂，孫 超，張來恩，陳文琴

(國(guó)家氣象信息中心，北京100081)

1 引言

氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)是能反映全國(guó)氣象業(yè)務(wù)實(shí)時(shí)運(yùn)行狀態(tài)的各種數(shù)據(jù)的集合[1-2]。為實(shí)現(xiàn)“全流程、全業(yè)務(wù)、一體化”的綜合業(yè)務(wù)監(jiān)視目標(biāo)，與一般的信息系統(tǒng)監(jiān)視數(shù)據(jù)相比，氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)涵蓋的內(nèi)容更廣泛，包含很多業(yè)務(wù)運(yùn)行的細(xì)節(jié)數(shù)據(jù)，數(shù)據(jù)量非常大，處理邏輯較為復(fù)雜，時(shí)效性要求高。

氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)可分為資源類監(jiān)視數(shù)據(jù)和業(yè)務(wù)類監(jiān)視數(shù)據(jù)。其中資源類監(jiān)視數(shù)據(jù)是從信息系統(tǒng)層面體現(xiàn)氣象業(yè)務(wù)軟硬件資源可用性，包括主機(jī)、存儲(chǔ)、網(wǎng)絡(luò)、安全、動(dòng)力環(huán)境等基礎(chǔ)設(shè)施以及數(shù)據(jù)庫、消息中間件等通用系統(tǒng)軟件、氣象業(yè)務(wù)應(yīng)用進(jìn)程運(yùn)行狀態(tài)數(shù)據(jù)；而業(yè)務(wù)類監(jiān)視數(shù)據(jù)是從業(yè)務(wù)層面體現(xiàn)氣象業(yè)務(wù)運(yùn)行狀態(tài)的監(jiān)視數(shù)據(jù)，在這類監(jiān)視數(shù)據(jù)中，一般包含具有氣象業(yè)務(wù)涵義的數(shù)據(jù)內(nèi)容，如收集的觀測(cè)資料到報(bào)率、可用率等[3]，與氣象業(yè)務(wù)流程和業(yè)務(wù)邏輯息息相關(guān)，能反映業(yè)務(wù)可用性等情況。

根據(jù)監(jiān)視數(shù)據(jù)的形態(tài)，氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)可細(xì)分為資源性能指標(biāo)類數(shù)據(jù)、原始業(yè)務(wù)日志數(shù)據(jù)、業(yè)務(wù)指標(biāo)類數(shù)據(jù)、告警事件類數(shù)據(jù)等。

1)資源性能指標(biāo)類數(shù)據(jù)是以指標(biāo)數(shù)值來反映資源的運(yùn)行能力狀態(tài)的數(shù)據(jù)，如存儲(chǔ)可用率、網(wǎng)絡(luò)流量等。

2)原始業(yè)務(wù)日志數(shù)據(jù)是業(yè)務(wù)各環(huán)節(jié)產(chǎn)生的日志數(shù)據(jù)，是主要的業(yè)務(wù)監(jiān)視數(shù)據(jù)來源，其中最核心的是氣象行業(yè)內(nèi)各種觀探測(cè)資料從采集、傳輸、預(yù)處理、存儲(chǔ)到產(chǎn)品加工、服務(wù)應(yīng)用的氣象資料全生命周期各環(huán)節(jié)的運(yùn)行狀態(tài)日志數(shù)據(jù)。除此之外，還包括各個(gè)氣象業(yè)務(wù)系統(tǒng)生成的業(yè)務(wù)日志數(shù)據(jù)，如定時(shí)任務(wù)的執(zhí)行情況、用戶對(duì)具體業(yè)務(wù)服務(wù)的訪問日志等。目前行業(yè)內(nèi)對(duì)關(guān)鍵的原始業(yè)務(wù)日志數(shù)據(jù)的內(nèi)容和格式提出了規(guī)范性要求，以便于數(shù)據(jù)的處理和管理。

3)業(yè)務(wù)指標(biāo)類數(shù)據(jù)是原始業(yè)務(wù)日志經(jīng)處理、統(tǒng)計(jì)等計(jì)算邏輯生成的能精細(xì)化反映業(yè)務(wù)運(yùn)行的狀態(tài)和可用性的指標(biāo)數(shù)據(jù)，如地面逐小時(shí)觀測(cè)資料的到報(bào)完整率、及時(shí)率、可用率等。這些業(yè)務(wù)指標(biāo)數(shù)據(jù)，不僅能監(jiān)視保障業(yè)務(wù)的正常穩(wěn)定運(yùn)行，還能提供業(yè)務(wù)管理決策支持。

4)告警事件類數(shù)據(jù)是基于前三者數(shù)據(jù)，經(jīng)過閾值和狀態(tài)判斷等計(jì)算分析發(fā)現(xiàn)的異常告警事件數(shù)據(jù)，包括告警事件的來源、種類、異常詳細(xì)情況等內(nèi)容，生成的告警類數(shù)據(jù)還需要進(jìn)行進(jìn)一步的告警規(guī)則處理及業(yè)務(wù)關(guān)聯(lián)分析。

目前，每小時(shí)采集和處理的氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)已達(dá)到1億條以上，要實(shí)現(xiàn)對(duì)海量監(jiān)視數(shù)據(jù)的高效采集和處理分析，本文提出了基于大數(shù)據(jù)技術(shù)的監(jiān)視數(shù)據(jù)采集和處理框架，能實(shí)現(xiàn)對(duì)海量原始監(jiān)視數(shù)據(jù)的高效采集、實(shí)時(shí)處理分析，系統(tǒng)框架同時(shí)具備穩(wěn)定性、容錯(cuò)性和可擴(kuò)展性等特征。

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

系統(tǒng)框架分為三層，包括原始監(jiān)視數(shù)據(jù)采集層、緩沖層、處理層。處理后的資源性能指標(biāo)類數(shù)據(jù)、原始業(yè)務(wù)日志數(shù)據(jù)、業(yè)務(wù)指標(biāo)類數(shù)據(jù)、告警事件類數(shù)據(jù)等將存儲(chǔ)在不同的數(shù)據(jù)庫中，如圖1。

圖1 系統(tǒng)框架

1)采集層。實(shí)現(xiàn)氣象綜合業(yè)務(wù)監(jiān)視原始日志的高效接入，并對(duì)接入數(shù)據(jù)進(jìn)行清洗，保證后端指標(biāo)計(jì)算環(huán)節(jié)數(shù)據(jù)的準(zhǔn)確性。采用REST接口及Flume日志收集系統(tǒng)兩種方式實(shí)現(xiàn)監(jiān)視數(shù)據(jù)采集。

2)緩沖層。對(duì)采集的監(jiān)視數(shù)據(jù)進(jìn)行緩存，實(shí)現(xiàn)監(jiān)視數(shù)據(jù)采集與處理環(huán)節(jié)的松耦合，為處理環(huán)節(jié)提供有效數(shù)據(jù)源。采用Kafka消息隊(duì)列實(shí)現(xiàn)對(duì)監(jiān)視日志數(shù)據(jù)和指標(biāo)數(shù)據(jù)的緩存。

3)處理層。根據(jù)氣象綜合業(yè)務(wù)監(jiān)視需求，對(duì)原始監(jiān)視數(shù)據(jù)進(jìn)行預(yù)處理、指標(biāo)計(jì)算并生成告警事件。采用Spark Streaming實(shí)時(shí)流數(shù)據(jù)處理框架實(shí)現(xiàn)高并發(fā)的監(jiān)視數(shù)據(jù)處理。

對(duì)監(jiān)視信息采集和處理之后，根據(jù)不同特點(diǎn)的監(jiān)視數(shù)據(jù)類型，針對(duì)日志類、指標(biāo)類、告警類監(jiān)視數(shù)據(jù)分別設(shè)計(jì)了不同的存儲(chǔ)模型：業(yè)務(wù)日志數(shù)據(jù)存儲(chǔ)在基于索引技術(shù)的Elasticsearch數(shù)據(jù)庫[4]，方便日志詳情查詢；資源性能指標(biāo)和業(yè)務(wù)指標(biāo)類數(shù)據(jù)采用時(shí)序值方式存儲(chǔ)在Cassandra數(shù)據(jù)庫[5]；告警事件類數(shù)據(jù)存儲(chǔ)在MongoDB數(shù)據(jù)庫，熱點(diǎn)數(shù)據(jù)存儲(chǔ)在Redis內(nèi)存數(shù)據(jù)庫。由于存儲(chǔ)的設(shè)計(jì)不是本文重點(diǎn)，故存儲(chǔ)模型設(shè)計(jì)不在此詳述。

3 監(jiān)視信息采集設(shè)計(jì)

氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)的采集主要通過“拉取”和“推送”兩種方式，拉取方式主要通過Flume技術(shù)，部署采集Agent到客戶端，采集監(jiān)視數(shù)據(jù)；推送方式主要由業(yè)務(wù)系統(tǒng)在運(yùn)行過程中，主動(dòng)調(diào)用REST接口實(shí)時(shí)推送監(jiān)視數(shù)據(jù)。不同類型監(jiān)視數(shù)據(jù)的采集方式和技術(shù)見表1。

表1 監(jiān)視信息采集方式

3.1 REST接口采集

REST接口采集通過調(diào)用REST(Representational StateTransfer，表述性狀態(tài)轉(zhuǎn)移)接口[6]實(shí)現(xiàn)客戶端通過HTTP層向服務(wù)器端推送監(jiān)視數(shù)據(jù)。在氣象業(yè)務(wù)系統(tǒng)中，實(shí)時(shí)收集、解析、處理生成監(jiān)視所需的運(yùn)行狀態(tài)信息，并通過調(diào)用REST接口，實(shí)時(shí)發(fā)送監(jiān)視數(shù)據(jù)，其發(fā)送的RESTful接口，以國(guó)家級(jí)地面自動(dòng)站收集日志采集為例，數(shù)據(jù)格式如下：

{

"type"："RT.CTS.STATION.DI"，

"system"："CTS"，

"message"："國(guó)內(nèi)氣象通信系統(tǒng)"，

"occur_time"：1518256800000，

"receive_time"：1518256922000，

"fields"：{

"FILE_NAME_N"："Z_SURF_C_BESZ_20180210100000_O_AWS_FTM_PQC.BIN"，

"LENGTH"：1286，

"IIiii"："54511"，

"STATION_NAME"："北京南郊站"，

"DATA_TYPE"："A.0001.0044.R001"，

"PROCESS_LINK"："2"，

"DATA_TIME"："2018-02-10 10：00"

"PROCESS_TIME"："2018-02-10 10：01：10.000"，

"TRAN_TIME"："2018-02-10 10：01：15.000"

}

}

收集日志包含“臺(tái)站編號(hào)”、“資料類型”、“資料時(shí)次”、“業(yè)務(wù)環(huán)節(jié)”等內(nèi)容，其中臺(tái)站編號(hào)、資料類型、資料時(shí)次用于計(jì)算數(shù)據(jù)到報(bào)情況，其它信息用于日志詳情查詢。

3.2 Flume采集

Flume是一個(gè)高可用的，高可靠的，分布式的海量日志采集、聚合和傳輸?shù)南到y(tǒng)，本系統(tǒng)采用Flume-ng，其核心是Agent，主要由source、channel、sink三個(gè)組件組成[7-8]。

Flume采集框架，包括采集層和匯聚層。采集層Agent負(fù)責(zé)采集氣象業(yè)務(wù)系統(tǒng)的運(yùn)行日志，并進(jìn)行格式轉(zhuǎn)換及封裝，匯聚層Agent負(fù)責(zé)匯聚所有業(yè)務(wù)系統(tǒng)的監(jiān)視數(shù)據(jù)。匯聚層相當(dāng)于業(yè)務(wù)總線的作用，起到隔離、緩沖的作用，如圖2。

圖2 Flume采集框架

1)隔離。屏蔽采集層單個(gè)Agent調(diào)整，而不影響其它監(jiān)視數(shù)據(jù)采集。Agent調(diào)整包括客戶端IP變動(dòng)，業(yè)務(wù)日志格式變化等。

2)緩沖。緩存采集層傳輸過來的數(shù)據(jù)到File Channel，一方面可以保證采集層采集Agent的不間斷執(zhí)行，另一方面可減緩大量監(jiān)視數(shù)據(jù)傳輸?shù)姆逯祲毫Γ⒈ＷC服務(wù)端監(jiān)視數(shù)據(jù)收集處理程序異常時(shí)，不丟失采集到的業(yè)務(wù)監(jiān)視數(shù)據(jù)。

另外，在Agent設(shè)計(jì)中，主要有以下關(guān)鍵點(diǎn)：

1)雙Channel設(shè)計(jì)。當(dāng)堆積在Channel中的Events數(shù)小于閾值時(shí)，所有的Events被保存在MemoryChannel中，Sink從MemoryChannel中讀取數(shù)據(jù)；當(dāng)堆積在Channel中的Events數(shù)大于閾值時(shí)，多余的部分自動(dòng)存放在FileChannel中，Sink從FileChannel中讀取數(shù)據(jù)。雙Channel可以充分使用MemoryChannel的高吞吐特性保證數(shù)據(jù)處理時(shí)效性，同時(shí)，利用FileChannel的緩存特性保證數(shù)據(jù)完整性。

2)故障切換設(shè)計(jì)。在采集層Agent，F(xiàn)ailover Sink Processor維護(hù)一個(gè)由3個(gè)Sink組成的優(yōu)先級(jí)Sink組件列表，只要有一個(gè)Sink組件接收或發(fā)送Event不可用，Event就被自動(dòng)切換到下一個(gè)組件。

3)負(fù)載均衡設(shè)計(jì)。在匯聚層Agent，使用負(fù)載均衡機(jī)制LoadBalance。部署2個(gè)Agent，均衡處理采集層Agent發(fā)送的監(jiān)控?cái)?shù)據(jù)，在每個(gè)Agent中，設(shè)計(jì)4個(gè)Sink，可以實(shí)現(xiàn)處理的高并發(fā)。采集數(shù)據(jù)轉(zhuǎn)換復(fù)雜的環(huán)節(jié)通過增加Sink組件提高Agent的吞吐量。

4 監(jiān)視信息處理設(shè)計(jì)

4.1 監(jiān)視數(shù)據(jù)預(yù)處理

監(jiān)視數(shù)據(jù)預(yù)處理主要針對(duì)氣象原始業(yè)務(wù)日志數(shù)據(jù)，資源性能指標(biāo)類監(jiān)視數(shù)據(jù)比較規(guī)整，不需要進(jìn)行預(yù)處理。氣象原始業(yè)務(wù)日志數(shù)據(jù)包括14大類498個(gè)子類的氣象觀測(cè)資料及產(chǎn)品從臺(tái)站到省級(jí)、國(guó)家級(jí)的收集、分發(fā)、處理、存儲(chǔ)、歸檔、服務(wù)的全業(yè)務(wù)流程運(yùn)行信息[9]。原始業(yè)務(wù)日志數(shù)據(jù)存在信息不完整、數(shù)據(jù)重復(fù)等問題，需要在監(jiān)視指標(biāo)計(jì)算前對(duì)數(shù)據(jù)進(jìn)行標(biāo)準(zhǔn)化構(gòu)建和數(shù)據(jù)清洗。

4.1.1 數(shù)據(jù)標(biāo)準(zhǔn)化構(gòu)建

數(shù)據(jù)標(biāo)準(zhǔn)化構(gòu)建包括數(shù)據(jù)合法性檢查、時(shí)間標(biāo)準(zhǔn)化處理、業(yè)務(wù)信息關(guān)聯(lián)和日志信息豐富等，處理流程如圖3。

圖3 數(shù)據(jù)標(biāo)準(zhǔn)化構(gòu)建處理流程

1)數(shù)據(jù)合法性檢查。根據(jù)業(yè)務(wù)特征和資料特征等合法性規(guī)則檢查監(jiān)視數(shù)據(jù)是否有效，無效數(shù)據(jù)存入Kafka錯(cuò)誤數(shù)據(jù)隊(duì)列，入庫備查。

2)時(shí)間格式化處理。統(tǒng)一各業(yè)務(wù)環(huán)節(jié)監(jiān)視數(shù)據(jù)的時(shí)間格式，所有時(shí)間統(tǒng)一為世界時(shí)，并精確到毫秒。

3)業(yè)務(wù)信息關(guān)聯(lián)。關(guān)聯(lián)業(yè)務(wù)配置信息，例如，某一種氣象資料是否業(yè)務(wù)考核、具體考核的比例、閾值、及時(shí)時(shí)間等是該資料的業(yè)務(wù)配置信息，則在此步驟中將通過關(guān)聯(lián)業(yè)務(wù)配置信息來判斷該氣象觀測(cè)資料是否業(yè)務(wù)考核資料，如果是業(yè)務(wù)考核，則獲取相關(guān)的考核信息。

4)日志信息豐富。根據(jù)原始業(yè)務(wù)日志中的臺(tái)站編號(hào)，從臺(tái)站信息庫中補(bǔ)充臺(tái)站所在的國(guó)家、區(qū)域/省、市、地縣等信息。

4.1.2 數(shù)據(jù)去重

為保證業(yè)務(wù)質(zhì)量，氣象資料在各業(yè)務(wù)環(huán)節(jié)存在數(shù)據(jù)訂正、數(shù)據(jù)傳輸續(xù)傳等現(xiàn)象，導(dǎo)致原始業(yè)務(wù)日志數(shù)據(jù)也存在重復(fù)問題。為保證后續(xù)監(jiān)視指標(biāo)計(jì)算的準(zhǔn)確性，預(yù)處理過程需要對(duì)數(shù)據(jù)進(jìn)行清洗去重操作，避免因重復(fù)計(jì)算造成誤差。

去重過程針對(duì)每一條原始業(yè)務(wù)日志，提取“資料類型”、“資料時(shí)次”、“業(yè)務(wù)環(huán)節(jié)”、“臺(tái)站號(hào)/文件”等要素作為唯一標(biāo)識(shí)，根據(jù)唯一標(biāo)識(shí)中的信息，計(jì)算其MD5值存入Redis緩存，后接入數(shù)據(jù)時(shí)檢查MD5值是否出現(xiàn)，若是則打上重復(fù)標(biāo)記。根據(jù)氣象資料業(yè)務(wù)時(shí)次周期處理的特點(diǎn)，在Redis中緩存4個(gè)數(shù)據(jù)周期，能夠覆蓋大部分?jǐn)?shù)據(jù)，對(duì)于少數(shù)超過4個(gè)周期的延遲數(shù)據(jù)，系統(tǒng)采用離線計(jì)算方法糾正。

首先對(duì)資料類型、資料時(shí)次、資料臺(tái)站和業(yè)務(wù)環(huán)節(jié)全排列編碼，計(jì)算方法如式(1)。

編碼值i=MD5(typei×timei×stationi×processi)

(1)

其中：type為資料小類編號(hào)；time為業(yè)務(wù)時(shí)次編號(hào)；station為臺(tái)站號(hào)；process為業(yè)務(wù)環(huán)節(jié)編號(hào)。

以500種資料小類，20種不同業(yè)務(wù)時(shí)次，10萬個(gè)臺(tái)站，10個(gè)處理環(huán)節(jié)為平均數(shù)估算最大占位數(shù)，如式(2)。

500×20×105×10=1010

(2)

在Redis中設(shè)置1010的Bitmap，占用內(nèi)存1010÷8=1.25×109字節(jié)，約為1.16GB。將每條資料日志記錄的編碼值i存入Redis中緩存，后續(xù)的日志記錄均與Redis緩存的編碼值進(jìn)行比較，如果相同，則打上重復(fù)標(biāo)記。

4.2 監(jiān)視指標(biāo)計(jì)算

原始業(yè)務(wù)日志數(shù)據(jù)預(yù)處理后經(jīng)過指標(biāo)計(jì)算處理，生成能精細(xì)化反映業(yè)務(wù)運(yùn)行的狀態(tài)和可用性的指標(biāo)數(shù)據(jù)，如完整率、到報(bào)率、及時(shí)率等。監(jiān)視指標(biāo)計(jì)算環(huán)節(jié)采用Spark Streaming實(shí)時(shí)處理框架，對(duì)實(shí)時(shí)數(shù)據(jù)流進(jìn)行統(tǒng)計(jì)計(jì)算。

為實(shí)現(xiàn)監(jiān)視指標(biāo)的實(shí)時(shí)更新，指標(biāo)計(jì)算以15秒為窗口從上游獲取日志數(shù)據(jù)，統(tǒng)計(jì)分鐘、小時(shí)、日尺度的業(yè)務(wù)監(jiān)視指標(biāo)，以及全國(guó)和分省指標(biāo)，并存入指標(biāo)庫。

4.2.1 業(yè)務(wù)指標(biāo)計(jì)算

氣象業(yè)務(wù)系統(tǒng)對(duì)于考核數(shù)據(jù)每天會(huì)生成應(yīng)收數(shù)據(jù)清單(節(jié)目表)，包含數(shù)據(jù)類型、應(yīng)收站號(hào)、應(yīng)到時(shí)次、環(huán)節(jié)標(biāo)識(shí)等信息，當(dāng)有數(shù)據(jù)到達(dá)某個(gè)環(huán)節(jié)時(shí)該環(huán)節(jié)會(huì)形成數(shù)據(jù)達(dá)到日志提交綜合業(yè)務(wù)監(jiān)視，通過與節(jié)目表的算法匹配計(jì)算出業(yè)務(wù)指標(biāo)，具體流程如下，以區(qū)域站數(shù)據(jù)小時(shí)到報(bào)率計(jì)算為例：

1)通過采集層獲取每日節(jié)目表，解析后將類型、站號(hào)、時(shí)次等信息寫入存儲(chǔ)層中。

2)從實(shí)時(shí)數(shù)據(jù)隊(duì)列中每15秒獲取一次日志數(shù)據(jù)，通過數(shù)據(jù)類型和環(huán)節(jié)標(biāo)識(shí)區(qū)分出區(qū)域站數(shù)據(jù)，并從存儲(chǔ)層中獲取區(qū)域站數(shù)據(jù)的節(jié)目表信息。

3)將收集環(huán)節(jié)的區(qū)域站日志中的站號(hào)、時(shí)次信息與進(jìn)行匹配節(jié)目表，計(jì)算15秒內(nèi)日志中區(qū)域站數(shù)據(jù)的收集情況。

4)從存儲(chǔ)層中獲取區(qū)域站到報(bào)指標(biāo)，根據(jù)步驟3)中的計(jì)算結(jié)果進(jìn)行累積匯總，計(jì)算出實(shí)時(shí)指標(biāo)回寫至存儲(chǔ)層，供綜合業(yè)務(wù)監(jiān)視前端展示調(diào)用。如圖4所示。

圖4 業(yè)務(wù)指標(biāo)計(jì)算流程

每次計(jì)算當(dāng)前時(shí)次、前兩個(gè)時(shí)次和后一個(gè)時(shí)次的到報(bào)情況，既能保證數(shù)據(jù)計(jì)算時(shí)效性，前端能取到實(shí)時(shí)的到報(bào)情況，又能保證計(jì)算資源的合理利用。由于氣象業(yè)務(wù)數(shù)據(jù)存在延遲到達(dá)的現(xiàn)象，指標(biāo)計(jì)算每天定時(shí)從存儲(chǔ)層中將近三天的數(shù)據(jù)取出進(jìn)行一次指標(biāo)重算，保證指標(biāo)庫中非實(shí)時(shí)數(shù)據(jù)的正確性。

4.2.2 計(jì)算任務(wù)容錯(cuò)處理

由于指標(biāo)計(jì)算時(shí)偶爾會(huì)出現(xiàn)計(jì)算失敗導(dǎo)致最終指標(biāo)計(jì)算結(jié)果存在誤差，系統(tǒng)設(shè)計(jì)了數(shù)據(jù)處理容錯(cuò)機(jī)制。采用批序號(hào)的增量計(jì)算方式，通過Redis緩存記錄當(dāng)前處理的批號(hào)，在指標(biāo)累計(jì)過程中比較最近的有效計(jì)算結(jié)果實(shí)現(xiàn)增量累計(jì)，只有在所有數(shù)據(jù)執(zhí)行計(jì)算完成之后才提交Kafka進(jìn)行消費(fèi)量偏移，若是計(jì)算失敗則納入下一批次重新計(jì)算。數(shù)據(jù)處理容錯(cuò)流程如圖5。

圖5 指標(biāo)計(jì)算容錯(cuò)處理流程

4.3 告警事件生成

根據(jù)業(yè)務(wù)規(guī)則，對(duì)監(jiān)視指標(biāo)進(jìn)行判斷和分析將生成告警事件，告警事件包括來源、種類、異常詳細(xì)情況等內(nèi)容。由于氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)包括業(yè)務(wù)全流程環(huán)節(jié)監(jiān)視指標(biāo)、以及支撐每個(gè)業(yè)務(wù)環(huán)節(jié)的基礎(chǔ)設(shè)施資源運(yùn)行狀態(tài)信息，一個(gè)環(huán)節(jié)出現(xiàn)告警，將導(dǎo)致后續(xù)業(yè)務(wù)流程均產(chǎn)生告警，而大批量的告警將影響運(yùn)維人員對(duì)關(guān)鍵告警的判斷。

減少告警數(shù)量、分析告警源頭，生成面向運(yùn)維人員的告警事件，將提高故障處理效率。告警歸因分析需要比較復(fù)雜的分析處理過程，還需關(guān)聯(lián)業(yè)務(wù)配置信息進(jìn)行綜合判斷。本系統(tǒng)主要實(shí)現(xiàn)了以下基礎(chǔ)的告警處理：

1)告警歸并。對(duì)同一種資料、同一個(gè)業(yè)務(wù)時(shí)次、不同業(yè)務(wù)環(huán)節(jié)的告警事件進(jìn)行歸并，只保留最前業(yè)務(wù)環(huán)節(jié)的告警事件。告警歸并方式包括有根據(jù)同一告警源頭進(jìn)行歸并處理，根據(jù)同一時(shí)間窗口內(nèi)的相似告警的歸并處理等，根據(jù)不同的業(yè)務(wù)需求采用不同的歸并處理算法，并對(duì)算法進(jìn)行優(yōu)化處理。

2)告警壓縮。設(shè)置壓縮規(guī)則，將同一種告警源頭、同一個(gè)業(yè)務(wù)時(shí)次、同一業(yè)務(wù)環(huán)節(jié)超時(shí)未處理的重復(fù)告警事件設(shè)置壓縮標(biāo)識(shí)，對(duì)重復(fù)事件進(jìn)行壓縮，并標(biāo)識(shí)該告警事件對(duì)應(yīng)的第一次和最后一次發(fā)生時(shí)間，壓縮所采用的算法包括有：基于滑動(dòng)窗口的壓縮方法，基于告警事件文本相似度的壓縮法，基于告警發(fā)生事件關(guān)聯(lián)關(guān)系的壓縮法等。對(duì)于某個(gè)具體業(yè)務(wù)應(yīng)用，可以配置壓縮時(shí)間窗口的大小，壓縮文本相似度大小，壓縮所需要的特定字段等

為了計(jì)算告警事件之間的關(guān)聯(lián)關(guān)系，首先需要對(duì)同一個(gè)時(shí)間窗口T內(nèi)的告警進(jìn)行去重處理，得到唯一個(gè)告警事件ID，這一個(gè)時(shí)間窗口T內(nèi)的所有獨(dú)立告警可以被認(rèn)為是一個(gè)獨(dú)立的數(shù)據(jù)集，即

AT={a0，a1，a2， …an}

其中AT表示時(shí)間窗口T內(nèi)所有告警的集合，而a0表示某一條獨(dú)立的告警事件。

計(jì)算告警事件之間的關(guān)聯(lián)規(guī)則就是形成如X→Y的映射關(guān)系，其中X和Y是不相交的項(xiàng)集，在關(guān)聯(lián)關(guān)系挖掘中，有兩個(gè)非常重要的概念：支持度(s)和置信度(c)，對(duì)于告警事件的支持度和置信度的計(jì)算公式分別為

其中：N是所有AT的總數(shù)。σ(X)表示X項(xiàng)集的個(gè)數(shù)。

3)告警升級(jí)。定義告警級(jí)別升級(jí)策略，對(duì)告警事件持續(xù)時(shí)間較長(zhǎng)而未處理關(guān)閉的告警進(jìn)行級(jí)別自動(dòng)升級(jí)處理，如從一般告警升級(jí)為嚴(yán)重告警；也可以依據(jù)告警事件發(fā)生頻次做升級(jí)處理，如對(duì)于發(fā)生頻次較高的告警做升級(jí)處理；或?qū)χ付ǜ婢愋瓦M(jìn)行升級(jí)處理，如對(duì)硬件類資源監(jiān)控指標(biāo)得到的告警事件進(jìn)行升級(jí)處理等。這些升級(jí)處理的觸發(fā)時(shí)間和觸發(fā)條件等可以通過配置來實(shí)現(xiàn)。

4.4 基于spark streaming的處理框架

為提高對(duì)海量氣象業(yè)務(wù)監(jiān)視數(shù)據(jù)的處理時(shí)效，系統(tǒng)采用Spark Streaming處理框架實(shí)現(xiàn)對(duì)監(jiān)視數(shù)據(jù)預(yù)處理、指標(biāo)計(jì)算。

Spark Streaming是將流式計(jì)算分解成一系列短小的批處理作業(yè)。這里的批處理引擎是Spark Core，也就是把Spark Streaming的輸入數(shù)據(jù)按照時(shí)間窗口分成一段一段的數(shù)據(jù)DStream(Discretized Stream)，每一段數(shù)據(jù)都轉(zhuǎn)換成Spark中的RDD(Resilient Distributed Dataset)，然后將Spark Streaming中對(duì)DStream的Transformation操作變?yōu)獒槍?duì)Spark中對(duì)RDD的Transformation操作，將RDD經(jīng)過操作變成中間結(jié)果保存在內(nèi)存中。整個(gè)流式計(jì)算根據(jù)業(yè)務(wù)的需求可以對(duì)中間的結(jié)果進(jìn)行疊加或者存儲(chǔ)到外部設(shè)備。Spark Streaming處理流程如圖6[10]。

圖6 Spark Streaming處理流程

對(duì)于一個(gè)Spark集群來說，執(zhí)行器(Executor)數(shù)量和執(zhí)行器的并行度固定后，分區(qū)數(shù)量決定了任務(wù)執(zhí)行的數(shù)量，也決定了任務(wù)執(zhí)行的效率。

針對(duì)執(zhí)行器數(shù)量，分區(qū)過多會(huì)導(dǎo)致任務(wù)的排隊(duì)，分區(qū)過少會(huì)降低集群的利用率；分區(qū)數(shù)據(jù)的傾斜，會(huì)導(dǎo)致各個(gè)任務(wù)的執(zhí)行時(shí)間不一致，快的任務(wù)需要等待慢的任務(wù)，形成木桶效應(yīng)，降低利用率。另一方面時(shí)間窗口大，每一批的數(shù)據(jù)量過大，會(huì)導(dǎo)致數(shù)據(jù)溢寫，降低效率，處理時(shí)間變長(zhǎng)；窗口小，批次數(shù)據(jù)過少，又會(huì)導(dǎo)致調(diào)度時(shí)間相對(duì)過長(zhǎng)。所以，分區(qū)和時(shí)間窗口的設(shè)置需要充分考慮系統(tǒng)資源情況和處理數(shù)據(jù)量。

氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)平均每秒處理記錄數(shù)約為3萬條，設(shè)置時(shí)間窗口為15s，每批數(shù)據(jù)量為45萬條；執(zhí)行器數(shù)量為12，故選擇把數(shù)據(jù)分成10個(gè)分區(qū)，對(duì)應(yīng)10個(gè)執(zhí)行器，另外2個(gè)執(zhí)行器進(jìn)行調(diào)度處理。一方面為了提高系統(tǒng)資源利用率，另一方面避免在計(jì)算過程中分區(qū)之間過多的數(shù)據(jù)交換。

由于計(jì)算中經(jīng)常會(huì)對(duì)同一種資料、同一個(gè)時(shí)次進(jìn)行指標(biāo)累加計(jì)算，如果將數(shù)據(jù)隨機(jī)分配到分區(qū)中執(zhí)行計(jì)算，同一種資料會(huì)在多個(gè)執(zhí)行器上計(jì)算，匯聚時(shí)需要從各個(gè)執(zhí)行器上進(jìn)行結(jié)果匯總，根據(jù)實(shí)驗(yàn)結(jié)果，影響了整體運(yùn)行效率。為了減少任務(wù)執(zhí)行后產(chǎn)生的匯聚開銷，在數(shù)據(jù)進(jìn)入分區(qū)前增加一次排序操作，按照資料類型和業(yè)務(wù)時(shí)次進(jìn)行排序，盡可能把同一種資料同一個(gè)時(shí)次的日志數(shù)據(jù)放在同一個(gè)執(zhí)行器上執(zhí)行。

把45萬條數(shù)據(jù)根據(jù)資料類型和時(shí)次均勻分成10個(gè)分區(qū)，數(shù)據(jù)分區(qū)的偽代碼如圖7。

圖7 數(shù)據(jù)分區(qū)偽代碼

為提高分區(qū)效率，本文采用水塘采樣算法，對(duì)每一個(gè)時(shí)間窗口的m條數(shù)據(jù)，隨機(jī)抽取n條樣本數(shù)據(jù)，針對(duì)樣本數(shù)據(jù)，按照分區(qū)邏輯計(jì)算每個(gè)分區(qū)的索引范圍，再將m條數(shù)據(jù)根據(jù)索引范圍加入到每個(gè)分區(qū)中。引入該方法后，數(shù)據(jù)分區(qū)效率得到大幅度提升。

根據(jù)復(fù)雜度[11]式(3)估算45萬條數(shù)據(jù)的排序性能。

計(jì)算復(fù)雜度=O(m)+n×log(n)+O(n)

(3)

分區(qū)優(yōu)化前，計(jì)算復(fù)雜度為

O(450000)+450000×log(450000)+

O(450000)≈3443946

(4)

分區(qū)優(yōu)化后，計(jì)算復(fù)雜度為

O(450000)+2000×log(2000)+O(2000)≈458602

(5)

可見，利用采樣數(shù)據(jù)進(jìn)行分區(qū)優(yōu)化，根據(jù)復(fù)雜度計(jì)算結(jié)果，分區(qū)性能可提升7倍以上。

4.5 基于全鏈路壓測(cè)體系的故障仿真與定位

基于全鏈路壓測(cè)體系的故障仿真建立了一套標(biāo)準(zhǔn)的模擬流程(圖8)，包含準(zhǔn)備階段、執(zhí)行階段、檢查階段和恢復(fù)階段。覆蓋從計(jì)劃到還原的完整過程，并通過可視化的方式清晰的呈現(xiàn)給使用者。故障模型訓(xùn)練的目的是有針對(duì)性的制造一些故障，給故障定位系統(tǒng)制造數(shù)據(jù)。其流程圖如圖9所示。

圖8 全鏈路壓測(cè)體系故障仿真流程圖

圖9 故障模型給故障定位系統(tǒng)制造數(shù)據(jù)流程圖

故障注入的對(duì)象包括1)應(yīng)用：外部超時(shí)/響應(yīng)變慢等。接口，DB延遲/連接滿/熱點(diǎn)，redis緩存熱點(diǎn)，kafka，中間件的負(fù)載均衡/限流等。2)資源：cpu，內(nèi)存，磁盤，io，網(wǎng)絡(luò)延時(shí)等。以及3)程序：cpu，mem，iptable，流量劫持，解析，過濾，模擬丟棄和延時(shí)等。

通過故障模擬，可以實(shí)現(xiàn)以下目標(biāo)：1)檢驗(yàn)降級(jí)預(yù)案的有效性：下游依賴出現(xiàn)故障時(shí)，預(yù)案能及時(shí)應(yīng)對(duì)，將系統(tǒng)的 SLA(service-level agreement，服務(wù)等級(jí)協(xié)議)維持在相對(duì)較高的水平，不因下游故障引起當(dāng)前服務(wù)可用性的故障。2)監(jiān)控報(bào)警：校驗(yàn)報(bào)警是否符合預(yù)期：是否報(bào)警、消息提示是否正確、報(bào)警的實(shí)效、收到報(bào)警的人是否預(yù)期。3)故障復(fù)現(xiàn)：故障復(fù)盤的后續(xù)todo項(xiàng)落地效果如何，通過一定時(shí)間后對(duì)故障的重現(xiàn)和驗(yàn)證，完成閉環(huán)。4)架構(gòu)容災(zāi)測(cè)試：主備切換、負(fù)載均衡、流量調(diào)度等容災(zāi)手段健壯性如何，提前發(fā)現(xiàn)并修復(fù)可避免的重大問題。5)參數(shù)調(diào)優(yōu)：限流策略、報(bào)警閾值、超時(shí)設(shè)置的調(diào)優(yōu)。

5 實(shí)驗(yàn)結(jié)果與分析

5.1 實(shí)驗(yàn)運(yùn)行環(huán)境

氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)采集與處理框架的實(shí)驗(yàn)運(yùn)行環(huán)境由11臺(tái)虛擬機(jī)組成，硬件配置如表2。

表2 硬件配置

5.2 性能分析

為測(cè)試監(jiān)視數(shù)據(jù)采集性能，選取了5個(gè)業(yè)務(wù)日志文件，計(jì)算Flume從讀取文件到寫入Kafka消息隊(duì)列之間的時(shí)效和吞吐量，測(cè)試結(jié)果表明，5個(gè)日志文件的平均處理吞吐量為18MB/s，每秒采集的日志記錄條數(shù)峰值達(dá)30萬以上，如圖10。

圖10 Flume采集性能

另外，基于Spark Streaming流處理平臺(tái)，接入所有氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)，測(cè)試24小時(shí)內(nèi)，預(yù)處理、去重、指標(biāo)計(jì)算環(huán)節(jié)的處理記錄條數(shù)，如圖11。

圖11 Spark Streaming處理性能

測(cè)試結(jié)果表明，Spark Streaming流處理平臺(tái)能實(shí)現(xiàn)平均每分鐘處理200萬條，每秒處理峰值達(dá)6萬條以上，日累計(jì)處理記錄數(shù)超過30億條。

5.3 告警壓縮結(jié)果

本文選取了2019年6月15日到6月20日共5天內(nèi)的告警事件數(shù)據(jù)，并采用上述告警壓縮方法，選定時(shí)間窗口為10分鐘，得到了壓縮前后的告警數(shù)量，如圖12所示。

圖12 對(duì)5天內(nèi)的告警數(shù)據(jù)做壓縮得到的結(jié)果。

圖12中展示了一部分時(shí)間內(nèi)的告警壓縮結(jié)果，其中紅色線表示壓縮之前的原始告警數(shù)量，而綠色線表示經(jīng)過告警壓縮之后的告警數(shù)量，從圖中可以對(duì)比發(fā)現(xiàn)，壓縮之后告警的數(shù)量得到了極大減少，且通過計(jì)算發(fā)現(xiàn)，該時(shí)間段內(nèi)的告警壓縮率結(jié)果為71.2%，表明經(jīng)過壓縮算法處理后，呈現(xiàn)出來的告警數(shù)量降低到原來的30%左右，顯示出壓縮算法的效果非常顯著。

5.4 故障仿真結(jié)果

為了驗(yàn)證系統(tǒng)的穩(wěn)定性和及時(shí)發(fā)現(xiàn)故障，解決故障的能力。從資源、應(yīng)用和程序三個(gè)方面進(jìn)行了故障注入的模擬測(cè)試。下面給出了其中兩個(gè)測(cè)試的實(shí)驗(yàn)結(jié)果。

1)進(jìn)程掛起故障模擬

圖13顯示了編號(hào)為1841的進(jìn)程掛起的故障模擬結(jié)果，故障注入之前狀態(tài)為S，故障注入之后的狀態(tài)變?yōu)門。

圖13 進(jìn)程掛起的故障模擬結(jié)果

該結(jié)果顯示故障注入系統(tǒng)可以成功的模擬和監(jiān)測(cè)進(jìn)程故障。

2)系統(tǒng)使用故障模擬注入

在該實(shí)驗(yàn)中，人為模擬了通過對(duì)系統(tǒng)進(jìn)行冒煙壓力測(cè)試的方式，加大系統(tǒng)的CPU使用率，圖14顯示了從12月29日到1月6日這一周之內(nèi)測(cè)試的結(jié)果。

圖14 測(cè)試結(jié)果

結(jié)果顯示，系統(tǒng)可以有效捕捉到CPU使用率超標(biāo)的系統(tǒng)緩慢故障并及時(shí)發(fā)出有效預(yù)警，給系統(tǒng)維護(hù)人員準(zhǔn)確判斷故障原因與及時(shí)排出故障提供了有效依據(jù)。

6 結(jié)束語

本文建立了針對(duì)氣象綜合業(yè)務(wù)監(jiān)視數(shù)據(jù)的采集與處理框架，采用REST、Flume、Kafka、Spark Streaming等一系列大數(shù)據(jù)技術(shù)與算法，實(shí)現(xiàn)對(duì)監(jiān)視數(shù)據(jù)的高效采集、緩沖和處理。同時(shí)采用先進(jìn)的故障模擬仿真技術(shù)對(duì)系統(tǒng)進(jìn)行了故障排查壓力測(cè)試。該框架已經(jīng)業(yè)務(wù)應(yīng)用于中國(guó)氣象局氣象綜合業(yè)務(wù)實(shí)時(shí)監(jiān)控系統(tǒng)中，目前，每日采集和處理數(shù)據(jù)超過30億條，每秒處理峰值達(dá)6萬條以上，支撐監(jiān)視數(shù)據(jù)30s內(nèi)完成采集、處理到告警的快速發(fā)布，提高了氣象綜合業(yè)務(wù)故障的快速發(fā)現(xiàn)能力。未來，基于該處理框架的監(jiān)視數(shù)據(jù)挖掘、告警關(guān)聯(lián)分析、故障智能預(yù)測(cè)和故障自愈將是研究的重點(diǎn)，以推動(dòng)氣象監(jiān)控運(yùn)維系統(tǒng)的自動(dòng)化、智能化水平。