基于Flink 復雜事件處理的空間站實驗柜排廢氣安全監測

施建明 王 偉 王 功

(中國科學院空間應用工程與技術中心, 北京 100094)

1 引言

空間站實驗柜的操作箱是航天員開展各種科學實驗、維護維修任務的場所,在每次任務完成后,往往會產生一定的廢氣,需要通過抽真空、充氮氣對箱體進行反復排空與沖洗。 操作箱采用密封設計,耐壓能力有限,與空間站艙內壓力形成一定的壓差,壓差在任何條件下必須保持在安全范圍內。 在抽真空、充氮氣工況下,操作箱體與艙內之間的壓差可能會超出箱體的標稱耐壓值甚至是耐壓極限,導致密封失效乃至箱體結構破壞,因此需要對排廢氣過程進行安全監測。

在排廢氣過程中,實時采集多種數據,包括箱體內外壓差、電磁閥開關狀態等,這些數據存在復雜的邏輯關系和時序關系,異常情景往往與復雜事件序列對應。 針對排廢氣過程較為復雜的安全監測問題,需要綜合考慮多個數據維度之間的耦合關系、數據流上下文信息以及時間信息等,簡單、靜態的分析方法難以應對此類問題。

復雜事件處理(Complex Event Processing,CEP)是一種針對實時數據流進行檢測、處理、分析和挖掘的技術方法。 何恒靖等[1]梳理了CEP技術在物聯網、分布式系統監控等領域的應用。Feng[2]介紹了集成了CEP 的大數據技術,提出將Spark Streaming 作為實現CEP 平臺的構想,并指出CEP 技術對于解決復雜物聯網系統狀態感知與實時決策問題的天然契合性。 Badr-Eddine等[3]提出了一種將CEP 和多種衛星遙感數據有效結合的衛星遙感數據處理和大氣污染預警的軟件架構,實現近于實時的大氣質量監測和預報。隨著大數據技術的發展,近年來作為第三代流式計算框架的Flink 越來越受到關注,已經成為大數據流式計算的事實標準。 Flink CEP 是在Flink上實現的復雜事件處理API 庫,常見的應用場景包括金融應用、欺詐檢測、復雜系統監控和報警[4]。 姬鴻飛[5]采用Flink CEP 技術,對可穿戴設備采集的多種心臟健康監測數據和運動加速度數據進行融合處理,實現預設規則的復雜事件匹配與檢測。

Flink 在多源傳感數據流并行處理、時間窗口智能推理計算等方面具有很高的性能,并能有效應對超時數據處理、故障容錯與恢復等問題,其在狀態監測和故障預警中的應用價值很高。 考慮到空間站實驗柜排廢氣安全監測的復雜性,本文提出將Flink CEP 應用到排廢氣過程監測中,并對該項技術在安全關鍵任務監測上的應用進行探索。

2 CEP 和Flink CEP

2.1 CEP 簡介

CEP 的概念是由Stanford 的Luckham 教授在20 世紀90 年代提出[6],是一種事件驅動的方法。CEP 是一種基于動態環境中事件流的分析技術,事件是指有意義的變量狀態變化。 通過分析事件間的關系,利用過濾、關聯、聚合等技術,根據事件間的時序關系和聚合關系制定檢測規則,持續地從數據流中查詢出符合要求的事件,最終分析得到更復雜的復合事件。 例如,發現溫度連續10 s上升后檢測到壓力超過10 MPa,輸出報警信號。

2.2 Flink CEP 庫與API

Flink CEP 是在Flink 之上實現的CEP 庫,使用它從無界數據流中檢測出復雜事件模式,以便掌握數據中的重要關鍵信息[7]。 在流處理程序開發環境中引入Flink CEP 依賴后,就可以使用該庫提供的Pattern API 編寫CEP 應用程序,首先要定義模式,然后是從數據流檢測出與模式匹配的事件序列。

2.2.1 定義模式

模式(Patterns)包括個體模式和模式序列。

個體模式是組成模式序列的單獨模式定義,可以是單例模式或者是循環模式。 所謂單例模式是指接受1 個事件的模式,而循環模式則接受超過1 個事件,將單例模式加上量詞即轉化成循環模式,例如“10 次溫度超過40 ℃”。

模式序列也叫組合模式,是由個體模式組合而成。 模式序列必須從1 個初始模式開始,然后就可以添加后續的模式,模式之間通過近鄰條件連接。 Flink CEP 支持3 種形式的事件近鄰形式:①嚴格近鄰;②寬松近鄰;③非確定近鄰。

在模式序列中,還可以設置時間約束,這往往是非常關鍵的,例如在初始模式匹配成功后,必須在10 s 內出現下一個事件。 這里的時間約束設置可以支持處理時間和事件時間。

2.2.2 檢測模式

在完成模式序列定義后,就要從輸入數據流中檢測出潛在的匹配。 給定輸入流input 和模式pattern,通過調用CEP.pattern(input, pattern)即可獲得1 個模式流(PatternStream)。 獲得PatternStream 之后,從中提取相應的事件序列,可以調用select/flatSelect API 來實現模式流的二次處理,生成用戶想要的結果。 在調用API 時需要傳入一個自定義的select function 函數作為參數,每個成功匹配的事件序列都會調用這個函數。

對于設置了時間約束的模式序列,如果初始模式匹配成功后,在檢測時間窗口內沒有匹配上后續事件,可以通過側輸出流處理超時數據。

3 排廢氣過程及安全模式

排廢氣是空間站實驗柜的安全關鍵任務,其過程包含較為復雜的測控操作,涉及多個系統之間的聯動,任何環節出現問題都會引起安全風險。

3.1 實驗柜密封箱體排廢氣

圖1 為實驗柜密封箱體排廢氣簡圖,圖中省略了部分閥門、管路細節,只保留了充排氣關鍵組件。 密封箱體經過幾輪的排廢氣、充氮氣循環,將箱體內的廢氣排凈。 排廢氣的流程如下:

圖1 密封箱體排廢氣簡化圖Fig.1 Simplified graph of EGD of sealed container

1) 打開排廢氣電磁閥,連通艙外真空環境,排廢氣持續一段時間,監測密封箱體與艙內的壓差dP,當dP超過-20 kPa 時,自動關閉排氣閥;

2)打開充氣閥,連通氮氣供應管路,向箱體充入氮氣并監測壓差,當壓差達到0 kPa 時,關閉充氣閥;

3)重復1)和2)若干輪,完成廢氣清洗任務。

密封箱體上安裝有壓差傳感器,用來測量密封箱體與艙內的壓差,為確保可靠,采用2 個相同的傳感器同時測量。 控制器根據壓差值,來控制排氣閥和充氣閥的開關。 采用兩級控制策略來控制閥門,其中排氣閥關閉的兩級壓差閾值分別為-20 kPa、-45 kPa,充氣閥關閉的兩級壓差閾值分別為0 kPa、45 kPa。 表1 為排廢氣過程安全監測變量列表。

表1 排廢氣安全監測相關變量Table 1 Variables monitored during EGD

3.2 排廢氣安全模式分析

按照第3.1 節的流程,參與排廢氣的全部設備均為正常狀態時,排廢氣過程能平穩安全地進行。 在發生可能導致密封箱體失壓或過壓的危險情況時,應能立即檢測到并進入安全模式,因此排廢氣過程需要嚴密監測以下幾種異常情景:

1)兩個壓差傳感器讀數不一致(相差超過0.2 kPa),關閉充氣閥和排氣閥,傳感器校準后再執行排廢氣任務。 由于采用壓差傳感器作為兩個電磁閥門閉環控制的輸入,因此為確保安全,傳感器一旦異常即進入安全模式。

2)排廢氣閥門打開,充氮氣閥門關閉,箱體壓差連續上升至-20 kPa,如果此時排氣閥未關閉,那么箱體壓差繼續上升至-45 kPa,此時如果二級控制仍未將排氣閥關閉,立即開啟充氣閥,箱體壓差開始逐步減小,并同時緊急關閉實驗柜外部的排廢氣管路上的閥門。

3)排廢氣閥門關閉,充氮氣閥門開啟,箱體壓力連續上升,直至壓差達到0 kPa,如果此時充氣閥未關閉,那么箱體壓力繼續上升至壓差值達到45 kPa,如果此時充氣閥仍未關閉,立即關閉氮氣供應源。

4 Flink CEP 監測程序

基于Flink 及其CEP 庫,設計并開發監測程序,實時監測排廢氣過程是否進入上文提出的3種安全模式,并輸出故障報警。

4.1 CEP 監測程序設計

4.1.1 程序流程設計

如圖2 所示,CEP 監測程序主要包括5 個步驟。

圖2 CEP 監測程序流程Fig.2 Monitoring program flow by CEP

1)讀取數據流。 Flink CEP 監測程序從外部系統讀取排廢氣過程產生的實時數據,在排廢氣安全監測場景下,需要至少秒級的數據采集頻次,可以通過Kafka 消息隊列來提供監測數據流的源。 Flink 通過與Kafka 的連接器讀取數據流,兩者配合能滿足狀態監測對精確一次一致性的要求。 從Kafka 中讀取數據,將原始數據轉化為樣例類(Case Class)“EgdData”格式的輸入數據流InputStream,便于后續數據處理。

2)時間特性設置和時間戳分配。 在Flink 流處理應用中,時間語義主要可以分為事件時間和處理時間。 表1 所列變量對應數據發生的時間(即事件時間),是排廢氣過程安全監測要考慮的重要信息,這些信息都包含在Kafka 生產者發布的數據源中。 首先,在環境中設置時間特性為事件時間。 然后,在完成數據流讀入操作后,調用“(assignAscendingTimestamps)”按照升序分配時間戳,這樣程序后續處理就能獲得InputStream 中所帶的時間信息。

3)CEP 匹配檢測。 將事先設計好的模式序列作用在InputStream 上,得到PatternStream。 針對3.2 節提出了排廢氣進入安全模式的3 個情景,開展3 次CEP 匹配檢測,輸出3 個不同的PatternStream。

4)匹配事件提取與報警輸出。 每個PatternStream 中都包含了與模式序列對應的匹配事件,需要從中提取出來報警信息,包括關鍵的事件及其對應的時間戳信息,并通過樣例類包裝成報警信息發出來。

5)超時數據處理與報警輸出。 當數據流中有超時數據,且模式序列中有時間約束條件時,通過側輸出流來處理超時情況,可以從PatternStream 中提取成功匹配的個體事件信息以及超時截止的時間戳,并通過樣例類包裝成超時報警信息發出來。

4.1.2 模式序列設計

針對第3.2 節的安全模式情景,設計3 種模式序列。

1)模式序列1(safteyPattern01)。 考慮到壓差傳感器可能因為干擾出現瞬態現象,這里將模式序列1 描述為:兩個壓差傳感器連續3 次出現相差過大后,排氣閥和充氣閥立即關閉,任務中止。

①初始事件的匹配條件為循環條件,滿足(dP1-dP2).abs＞0.2 kPa。

②結尾事件與初始事件之間為嚴格近鄰關系(next),后續事件匹配條件為(vS1 =="closed")&&(vS2=="closed") 。

2)模式序列2(safetyPattern02)。 模式序列2描述為:排氣閥打開、充氣閥關閉,經過一段時間抽氣后,2 個壓差傳感器均正常并且測量值均超過-20 kPa,隨后壓差繼續上升并超過-45 kPa,緊接著充氣閥打開,隨后壓差開始下降。

① 初始事件的匹配條件為((dP1-dP2).abs≤0.2 kPa) && ((dP1 ＜ - 20 kPa) &&(dP2 ＜- 20kPa)) 。

② 中間事件1 的匹配條件為((dP1-dP2).abs≤ 0.2kPa) && ((dP1 ＜ - 45kPa)&&(dP2 ＜- 45kPa)) 。

③ 中間事件2 的匹配條件為(vS1 =="open")&&(vS2=="open") 。

④ 結尾事件的匹配條件為((dP1-dP2).abs≤0.2 kPa) && ((dP1 ＞ - 45 kPa) &&(dP2 ＞- 45 kPa)) 。

3)模式序列3(safetyPattern03)。 模式序列3描述為:排氣閥關閉、充氣閥打開,經過一段時間充氣后,2 個壓差傳感器均正常并且測量值均超過0 kPa,隨后壓差繼續上升并超過45 kPa,緊隨其后充氣閥仍然為打開狀態。

① 初始事件的匹配條件為((dP1-dP2).abs≤0.2 kPa)&&((dP1 ＞0 kPa)&&(dP2 ＞0 kPa)) 。

② 中間事件的匹配條件為((dP1-dP2).abs≤0.2 kPa)&&((dP1 ＞45 kPa)&&(dP2 ＞45 kPa)) 。

③ 結尾事件的匹配條件為 ((vS1 =="open")&&(vS2=="closed")) 。

4.2 CEP 監測程序開發

根據第4.1.1 節的程序流程設計,開發CEP監測程序代碼,核心是編寫CEP 模式序列代碼和事件提取自定義函數類代碼,其他代碼塊均為Flink 流處理程序代碼。 這里基于Flink 框架和CEP 庫,采用Scala 語言開發排廢氣安全監測應用程序。

4.2.1 CEP 模式序列代碼

采用Flink CEP 庫提供的API 編寫模式序列,聲明3 個Pattern 類型的變量safetyPattern01/02/03,編寫相應的模式序列代碼,見表2。

①safetyPattern01 由2 個個體模式組合而成,通過嚴格近鄰next( )連接,每個模式的匹配循環條件分別為times(3)和times(2),并且添加了6 s的時間約束條件Time.seconds(6)。

②safetyPattern02 由4 個個體模式組合而成,前2 個通過寬松近鄰先后連接,后3 個通過嚴格近鄰先后連接。 每個個體模式都是多個變量滿足特定條件的組合,其中begin2 模式包含了循環條件times(3)。

③safetyPattern03 由3 個個體模式組合而成,前2 個通過寬松近鄰先后連接,后2 個通過嚴格近鄰先后連接。 每個模式的匹配條件均為循環條件,分別為times(3)、times(3)、times(2)。

在每個模式序列begin 后的[EgdData],代表輸入流的數據格式。 將編寫好的3 個模式序列分別作用于輸入流,即得到patternStream01/02/03 3個PatternStream。

4.2.2 事件提取與報警輸出函數類代碼

對patternStream01/02/03 分別執行事件提取與信息輸出操作,獲取到排廢氣過程安全監測可能出現的異常事件序列并輸出報警信息,這里通過繼承于PatternSelectFunction 的自定義函數類來定制開發,關鍵是重寫select 方法,這里需要針對每個PatternStream 進行方法的重寫,實現代碼見表3。

表3 重寫“select( )”方法代碼Table 3 Code of method “select( )” override

select 方法的輸入是1 個Map 變量,在經過模式檢測后,所有成功匹配的事件都存儲在該變量中。 Map 的key 是String 類型的模式名稱,即表2 中個體模式的名稱,例如begin1,middle2-1,end1 等;value 是類型為EgdData 的List,用于存儲匹配的數據點,這里之所以是List 是因此循環條件會匹配出多個事件。

表2 用于排廢氣安全監測的CEP 模式序列代碼Table 2 Code of CEP pattern sequence applied for EGD safety monitoring

select 方法的輸出是Warning01/02/03 樣例類格式的數據,分別用于包裝壓差傳感器異常、排氣閥門異常、充氣閥門異常3 種報警信息流。

1)在SafetyPattern01 Match 函數類中,select方法提取的是壓差傳感器首次出現異常的事件時間和結尾2 個閥門均關閉的事件時間,最終輸出包裝了這2 個時間數據和傳感器異常“sensory abnormal!”的報警信息。

2)在SafetyPattern02 Match 函數類中,select 方法提取的是4 個關鍵事件的時間,分別是負壓壓差超過第一閾值的時間、超過第二閾值的時間、閥門狀態匹配的時間、負壓壓差開始下降的時間,最后輸出包裝了這4 個時間數據和排氣閥門異常“ dischargevalve fails to close!”的報警信息。

3)在SafetyPattern03 Match 函數類中,select方法提取的是3 個關鍵事件的時間,分別是正壓壓差超過第一閾值的時間、超過第二閾值的時間、閥門狀態匹配的時間,最后輸出包裝了這3 個時間數據和充氣閥門異常“chargevalve fails to close!”的報警信息。

5 測試與討論

模擬排廢氣過程,涵蓋正常過程和3 種故障模式,監測數據的采集頻率為1 Hz。 通過測試檢驗Flink CEP 監測程序能否正確檢測出3 種故障模式,對測試性指標進行驗證,討論數據超時的處理。

5.1 排廢氣過程模擬與監測數據生成

按照如下流程,模擬排廢氣過程并生成相應的仿真數據用于測試:

1)第一次為正常排充氣循環,排廢氣開始時,密封箱體內外壓差為0 kPa,排氣閥打開,同時充氣閥處于關閉狀態。 假設此排氣階段壓差變化速率為55.6 Pa/s,經過6 min 后,壓差變為-20 kPa 左右。 緊接著,排氣閥關閉,充氣閥打開,開始充氣過程,假設此充氣階段壓差變化速率為111.2 Pa/s,經過3 min 后,壓差變為0 kPa左右。

2)第二次循環模擬壓差傳感器故障,排廢氣開始后2 min,模擬2 個壓差傳感器開始出現連續的不一致,并且1 s 后2 個閥門均變為關閉狀態。異常數據延續5 s,該循環終止。

3)第三次循環模擬排氣閥關不上故障,第一階段排廢氣過程和第一次循環一樣,當壓差超過-20 kPa 后,排氣閥未關閉,壓差經過4 min 繼續擴大至- 45 kPa,此階段壓差變化速率為104.2 Pa/s。 緊接著,充氣閥打開,壓差比上一時刻有所下降,此狀態持續5 s 后循環終止。

4)第四次循環模擬充氣閥關不上故障,充氣開始,此時排廢氣閥處于關閉狀態,充氣閥為打開狀態, 循環初始時密封箱體的內外壓差為-20 kPa,充氣過程與第1 次循環一樣,當壓差超過0 kPa 時,充氣閥不關閉,壓力經過8 min 后,壓差變為45 kPa,變化速率為93.8 Pa/s,最后連續5 s 充氣閥依然保持開狀態,并且壓差繼續上升,循環模擬終止。

基于以上排廢氣過程,生成對應表1 中變量的監測數據。 模擬排廢氣過程共計1935 s,生成數據1935 條。 在每秒采集的壓差數據中加入標準差為5 Pa 的隨機白噪聲,模擬數據的正常波動。

5.2 測試結果與分析

在Kafka 數據服務器上每秒逐條發布5.1 節的數據,作為測試用的數據流來源。 在本地開發環境測試Flink CEP 監測程序,通過控制臺打印輸出測試結果,分別測試3 種模式檢測結果的正確性以及報警生成時間與結尾事件時間的延遲。

如圖3 所示,CEP 監測程序成功匹配出數據流中的3 種模式序列,并輸出了相應的報警信號。圖3 中虛線將3 種模式序列的識別結果劃分開,不同顏色、不同形狀的圖形代表相應的匹配事件,并標記出相應的事件時間,最后的圖案代表報警信號發出,標記了報警信號生成的時間,時間語義為處理時間。

圖3 模式序列匹配測試結果Fig.3 Testing results of pattern sequences matching

1)安全模式1 處理結果分析:CEP 程序首先匹配出滿足兩個壓差傳感器連續3 次相差過大循環條件的事件,橙色圓形下標記第一條數據的事件時間;緊接著,匹配出充氣閥、排氣閥連續2 次均處于關閉狀態事件;最后發出Warning01 報警信號。 報警信號與匹配的結尾事件最后1 條數據的事件時間相差1222 ms。

2)安全模式2 處理結果分析:CEP 程序首先匹配出壓差超過-20 kPa 事件,綠色圓形下標記壓差穿越一級負壓閾值的事件時間;一段時間后,成功匹配出壓差超過-45 kPa 事件;緊接著,匹配出“充氣閥立即打開”事件;緊接著,匹配出壓差低于-45 kPa 事件。 最后發出Warning02 報警信號。 報警信號與結尾事件的事件時間相差1170 ms。

3)安全模式3 處理結果分析:CEP 程序首先匹配出壓差超過0 kPa 事件藍色圓形下標記壓差穿過一級正壓閾值的事件時間;一段時間后,成功匹配出壓差超過45 kPa 事件;緊接著,匹配出充氣閥連續2 次均處于打開狀態事件;最后發出Warning03 報警信號。 報警信號與結尾事件最后1 條數據的的事件時間相差1217 ms。

5.3 測試性指標驗證情況

1)故障檢測率與隔離率驗證。 增加循環模擬次數到12 次,擴充故障樣本量到9 次,測試故障檢測率和隔離率指標。 得到如表4 所示的測試結果。 模擬的9 次故障均能被程序檢測到,給出的診斷結果隔離到相應的部件故障模式,故障檢測率和隔離率均為100%。 故障診斷信息發出來的時刻與故障發生時刻之間的時間差均值為1921 ms。 測試時,仿真程序和檢測程序運行在本地環境,但仿真程序向遠程Kafka 集群發布數據,而檢測程序又從Kafka 集群拉取數據,因此表4中的差值包含了數據的網絡傳輸時間。

表4 故障檢測與隔離測試結果Table 4 Fault detection and isolation test results

2)虛警率測試與討論。 在壓差傳感器測量密封箱體內外壓差過程中,由于信號干擾等可能會在數據流中引入瞬時數據野點,安全監測程序應能抗干擾,以避免虛警產生。 為此,將測試數據的第127/289/980 條數據點中的dP2 或dP1 改成異常值,使dP2 與dP1 之間的差值超過0.2 kPa。由于CEP 采用了復雜的模式序列設計,該瞬時異常數據不符合模式序列的匹配規則,經實測對應異常點無報警信號輸出,在此測試條件下虛警率為0。 其他真實故障的報警信號輸出不受影響。

5.4 超時數據測試與討論

以安全模式1 為例來測試Flink CEP 對超時數據的處理,為了測試方便,截取傳感器異常發生時段前后的30 條數據,并將閥門關閉時的數據設置為2 s 延遲,模擬數據傳輸超時異常工況。

首先,測試無側輸出流的情況,由于后續事件序列已經超過了6 s 的時間約束條件,程序認為沒有匹配成功,未輸出任何報警信號。 然后,在CEP 程序中增加側輸出流用來處理超時情況,超時報警信息設計4 個字段:傳感器異常事件時間、“begin1”個體模式匹配數據條數、超時時間戳、超時報警字符串。 經測試,一共輸出5 條超時報警信息,見表5。 其中第1 個時間戳分別對應數據流中的5 個傳感器異常數據的事件時間,第2 個時間戳則是分別在第1 個時間戳的基礎上加6000 ms,表示6 s 內未成功匹配后續事件,直接給出超時報警提示。

6 結論

通過Flink CEP 在空間站實驗柜排廢氣安全監測上的應用研究,開發相應的監測應用程序,通過測試,得出以下結論:

1)Flink CEP 能準確識別出排廢氣過程中復雜的異常事件序列,能應對包含多維數據耦合和數據流上下文關聯的復雜事件處理問題;

2)監測程序輸出報警的處理時間與序列結尾事件時間相差在1～2 s 之間,能及時輸出報警信號;

3)監測程序具有抗干擾和處理超時數據的能力,適用于實際數據流監測場景。