王 耀， 趙 炯， 周奇才， 熊肖磊， 陳傳林， 張 恒

（1. 同濟(jì)大學(xué) 機(jī)械與能源工程學(xué)院，上海 201804；2. 同濟(jì)大學(xué) 浙江學(xué)院，浙江 嘉興 314051；3. 上海地鐵盾構(gòu)設(shè)備工程有限公司，上海 200233）

“工業(yè)4.0” 背景下，生產(chǎn)制造過程、工程裝備狀態(tài)監(jiān)測等環(huán)節(jié)產(chǎn)生了海量數(shù)據(jù)[1-2］。工業(yè)大數(shù)據(jù)的分析與應(yīng)用為企業(yè)的生產(chǎn)、管理，裝備的智能控制、故障診斷提供了可靠的決策依據(jù)[3-4］。工業(yè)數(shù)據(jù)的分析效果又與數(shù)據(jù)質(zhì)量緊密相關(guān)[5］，然而在工業(yè)環(huán)境中，采集的數(shù)據(jù)受環(huán)境、可靠性、通信等因素影響，缺失值、異常值、冗余等[6］臟數(shù)據(jù)已成為常態(tài)。使用含有異常的數(shù)據(jù)進(jìn)行數(shù)據(jù)分析、建模等應(yīng)用，會(huì)直接影響決策結(jié)果，導(dǎo)致模型失效[2，5］，因此數(shù)據(jù)在應(yīng)用前需進(jìn)行必要的清洗。而工業(yè)數(shù)據(jù)的清洗又獨(dú)具特點(diǎn)，工業(yè)大數(shù)據(jù)區(qū)別于傳統(tǒng)互聯(lián)網(wǎng)大數(shù)據(jù)，時(shí)序性、關(guān)聯(lián)性、周期性是工業(yè)大數(shù)據(jù)的顯著特點(diǎn)[7-8］，同時(shí)工業(yè)數(shù)據(jù)來源于實(shí)際的工程裝備，采集的數(shù)據(jù)具有物理含義。工業(yè)數(shù)據(jù)的時(shí)序性、關(guān)聯(lián)性已成為工業(yè)數(shù)據(jù)清洗系統(tǒng)設(shè)計(jì)的關(guān)鍵因素[9］，對(duì)實(shí)時(shí)性較高的控制系統(tǒng)，還要求數(shù)據(jù)可實(shí)時(shí)處理。雖然異常數(shù)據(jù)會(huì)導(dǎo)致模型的不準(zhǔn)確，但工業(yè)場景的異常值往往成為狀態(tài)檢測和健康評(píng)估的重要依據(jù)，如故障分析應(yīng)用中異常數(shù)據(jù)價(jià)值較高，數(shù)據(jù)清洗系統(tǒng)還應(yīng)當(dāng)為異常數(shù)據(jù)的保留或修復(fù)提供妥善的選擇。

目前，常見的工業(yè)數(shù)據(jù)清洗模式為離線清洗。離線清洗系統(tǒng)對(duì)實(shí)時(shí)數(shù)據(jù)先存儲(chǔ)、后清洗，如Kumar等人[10］針對(duì)數(shù)據(jù)庫設(shè)計(jì)的離線清洗系統(tǒng)，傳統(tǒng)的離線方法無法適用于較高實(shí)時(shí)性的工業(yè)環(huán)境。在新興分布式框架流行后，實(shí)時(shí)流式清洗系統(tǒng)逐漸成為工業(yè)數(shù)據(jù)處理的主流系統(tǒng)，陳志云[11］等人提出了一種基于Storm 的工業(yè)流水線實(shí)時(shí)分析系統(tǒng)，該系統(tǒng)采用Storm + Kafka + Flume + HBase的實(shí)時(shí)流處理框架，證實(shí)了此類系統(tǒng)具有較好的可靠性與并行度。Geng[12］等人也利用Flume，Kafka，Hbase等框架搭建了工業(yè)數(shù)據(jù)分析系統(tǒng)，并證實(shí)了主流分布式框架能滿足工業(yè)現(xiàn)場的數(shù)據(jù)采集、分析需求。

流式數(shù)據(jù)清洗系統(tǒng)通常集成了高精度、低偏差的清洗算法。工業(yè)時(shí)序數(shù)據(jù)的清洗算法通常以處理缺失值、錯(cuò)列、異常值等為目標(biāo)[13-14］。其中異常值又分為連續(xù)異常、單點(diǎn)異常、平移異常等類型[15］。在來源于工程裝備的時(shí)序數(shù)據(jù)中，單點(diǎn)異常是常見的一種。所謂單點(diǎn)異常，是指觀測值和真實(shí)值具有較大偏差，且這種大偏差在一組時(shí)序數(shù)據(jù)中不連續(xù)出現(xiàn)。單點(diǎn)異常是工業(yè)現(xiàn)場較普遍的類型，常用的清洗方法有①基于平滑方法的清洗。利用數(shù)據(jù)擬合去除時(shí)間序列中的異常值，如移動(dòng)平均法（SMA），加權(quán)平均法（WMA）等[16］。Jeffery 等人基于滑動(dòng)窗口法設(shè)計(jì)了在線數(shù)據(jù)流清洗系統(tǒng)[17］。通常平滑算法能夠在時(shí)序數(shù)據(jù)清洗中發(fā)揮良好的效果，但清洗準(zhǔn)確率較低，清洗時(shí)會(huì)影響部分正常數(shù)據(jù)導(dǎo)致整體清洗偏差增大。②基于約束的清洗。通過設(shè)計(jì)某種規(guī)則，調(diào)整異常數(shù)據(jù)使其滿足設(shè)定的約束。Zhang等人[9，13］提出了速度約束清洗算法，通過限制一個(gè)時(shí)間段內(nèi)的兩個(gè)點(diǎn)數(shù)值的變化率，從全局或局部識(shí)別違反約束的點(diǎn)，利用異常點(diǎn)前后時(shí)刻的值對(duì)異常點(diǎn)修復(fù)。Song 等人[18］在速度約束研究的基礎(chǔ)上擴(kuò)展了基于加速度約束的數(shù)據(jù)清洗方法。這種基于速度約束的清洗方法處理單點(diǎn)異常值具有很好的效果。但該方法在處理周期性數(shù)據(jù)時(shí)或周期跳變數(shù)據(jù)時(shí)產(chǎn)生數(shù)據(jù)失真，無法適用強(qiáng)周期性工業(yè)數(shù)據(jù)的清洗需求。③基于機(jī)器學(xué)習(xí)的清洗。通過構(gòu)建聚類、回歸、神經(jīng)網(wǎng)絡(luò)等模型識(shí)別并修復(fù)異常值[19-20］，機(jī)器學(xué)習(xí)的清洗算法雖然高效，但數(shù)據(jù)集中異常值偏少時(shí)，類別不均衡，對(duì)標(biāo)簽集的修復(fù)結(jié)果受樣本特征影響，含有較多異常值的樣本集會(huì)降低修復(fù)質(zhì)量。

本文為了高效清洗具有時(shí)序性、周期性等特點(diǎn)的工業(yè)數(shù)據(jù)，設(shè)計(jì)了一套面向周期性工業(yè)時(shí)序數(shù)據(jù)的流式清洗系統(tǒng)和清洗算法，使系統(tǒng)在處理周期性突變數(shù)據(jù)時(shí)，也具有良好的數(shù)據(jù)修復(fù)效果。首先，利用Flume、Kafka 等組件，構(gòu)建一套流式數(shù)據(jù)處理系統(tǒng)，克服傳統(tǒng)離線清洗系統(tǒng)先存儲(chǔ)、后清洗的實(shí)時(shí)性差問題；其次，在構(gòu)建的流式清洗系統(tǒng)基礎(chǔ)上，使用基于速度約束的清洗方法[9，13］，并綜合數(shù)據(jù)的時(shí)序性、周期性、物理意義，對(duì)速度約束模型做周期性數(shù)據(jù)的適應(yīng)性優(yōu)化，解決傳統(tǒng)算法在清洗距離、準(zhǔn)確度方面的不足，提升速度約束算法在周期性工業(yè)數(shù)據(jù)的適應(yīng)性。最后通過試驗(yàn)驗(yàn)證所提周期性數(shù)據(jù)清洗算法的先進(jìn)性。

1 工業(yè)時(shí)序數(shù)據(jù)的流式清洗系統(tǒng)架構(gòu)

為滿足工業(yè)數(shù)據(jù)先清洗、再存儲(chǔ)的實(shí)時(shí)處理需求，本文使用主流的大數(shù)據(jù)框架，搭建了一套用于工業(yè)時(shí)序數(shù)據(jù)清洗、存儲(chǔ)的流式處理系統(tǒng)。整套系統(tǒng)包括數(shù)據(jù)匯集組件，連接組件，緩存組件，清洗組件，存儲(chǔ)組件，系統(tǒng)框架如圖1所示。

圖1 清洗系統(tǒng)組件和架構(gòu)Fig.1 Component and architecture in the cleaning system

數(shù)據(jù)匯集組件選用Mosquitto MQTT Broker，其位于系統(tǒng)下層，使用MQTT協(xié)議實(shí)現(xiàn)底層采集裝置和數(shù)據(jù)清洗系統(tǒng)的通信。采集的數(shù)據(jù)須包含時(shí)間戳以及至少一列的有效觀測數(shù)據(jù)。Mosquitto匯集并封裝采集的數(shù)據(jù)，隨后數(shù)據(jù)被發(fā)往數(shù)據(jù)連接組件進(jìn)行后續(xù)流轉(zhuǎn)。

數(shù)據(jù)連接組件選用Flume框架，其在清洗系統(tǒng)中的位置和業(yè)務(wù)邏輯如圖1所示。匯集的數(shù)據(jù)由MQTT Source轉(zhuǎn)入Interceptor組件做簡單過濾，修復(fù)明顯的異常值，并將初步清洗后的數(shù)據(jù)嵌入PreCleaned標(biāo)簽，由Sink送入緩存組件。過濾時(shí)任何不符合格式的數(shù)據(jù)、類型錯(cuò)誤數(shù)據(jù)被嵌入Error標(biāo)簽直接進(jìn)入存儲(chǔ)組件。使用Memory Channel和多路復(fù)用結(jié)構(gòu)，不同標(biāo)簽的數(shù)據(jù)分別發(fā)往Kafka進(jìn)行清洗或直接存儲(chǔ)。

數(shù)據(jù)緩存組件由Kafka集群組成，用于待清洗數(shù)據(jù)的緩存和系統(tǒng)解耦，適配高速產(chǎn)生的數(shù)據(jù)和清洗組件較慢的處理速率。圖1右側(cè)展示了其業(yè)務(wù)邏輯，每個(gè)數(shù)據(jù)采集裝置使用一個(gè)Topic，每個(gè)Topic包含一組特定的時(shí)間序列數(shù)據(jù)。Producer為Flume組件的Sink端，用于將可用數(shù)據(jù)轉(zhuǎn)入到緩存組件。Consumer為數(shù)據(jù)清洗組件，其周期性地從Kafka消費(fèi)數(shù)據(jù)，完成對(duì)數(shù)據(jù)的清洗工作，結(jié)果寫入數(shù)據(jù)存儲(chǔ)組件。

數(shù)據(jù)清洗組件是數(shù)據(jù)清洗系統(tǒng)的核心部分。清洗組件從Kafka持續(xù)獲取指定Topic下的數(shù)據(jù)，完成基于速度約束的數(shù)據(jù)清洗工作。消費(fèi)者每隔一定間隔從Kafka消費(fèi)數(shù)據(jù)，依據(jù)數(shù)據(jù)格式進(jìn)行預(yù)處理，把原始數(shù)據(jù)封裝成可清洗數(shù)據(jù)集；隨后經(jīng)過處理算法完成異常值檢測和異常數(shù)據(jù)修復(fù)工作。

數(shù)據(jù)存儲(chǔ)組件由HDFS、Hive、HBase 構(gòu)成。清洗組件依據(jù)Flume 連接階段數(shù)據(jù)標(biāo)簽的差異，調(diào)用不同API寫入存儲(chǔ)組件。清洗系統(tǒng)修復(fù)所有不滿足約束的異常數(shù)據(jù)，并存儲(chǔ)修復(fù)結(jié)果。但被清洗組件檢測為異常的數(shù)據(jù)以及被連接組件標(biāo)記為Error 的數(shù)據(jù)，會(huì)被冗余存儲(chǔ)，方便應(yīng)用于故障檢測等工業(yè)應(yīng)用；存儲(chǔ)組件同時(shí)存儲(chǔ)修復(fù)前異常值和修復(fù)后結(jié)果，為數(shù)據(jù)清洗后的數(shù)據(jù)分析工作提供保障。

清洗系統(tǒng)由上述組件構(gòu)成，作為清洗算法的載體，能夠?qū)崿F(xiàn)海量數(shù)據(jù)的流式處理和清洗任務(wù)，相比傳統(tǒng)清洗系統(tǒng)具有低耦合、可擴(kuò)展、易維護(hù)、大緩沖等特點(diǎn)。

2 面向周期性時(shí)序數(shù)據(jù)的速度約束清洗方法

首先給出速度約束的概念：一組時(shí)序數(shù)據(jù)中，若兩個(gè)時(shí)序點(diǎn)數(shù)值的變化率限制在一定范圍內(nèi)，則稱兩點(diǎn)滿足速度約束。圖2a展示了一組時(shí)間序列，設(shè)時(shí)間序列X=x[t1］，x[t2］，x[t3］，……其中ti為時(shí)間戳，x[ti］是時(shí)間為ti時(shí)的數(shù)值，簡記為xi。

圖2 時(shí)間序列和速度約束示意[9］Fig.2 Time series and speed constraints

考察圖2a 中左側(cè)時(shí)間間隔為w的兩個(gè)數(shù)據(jù)點(diǎn)，若兩個(gè)數(shù)據(jù)點(diǎn)的值xn，xm之間滿足式（1），則稱兩個(gè)數(shù)據(jù)點(diǎn)在窗口w內(nèi)滿足速度約束s= （smin，smax）。由于機(jī)械結(jié)構(gòu)或運(yùn)動(dòng)構(gòu)件具有慣性，正常工作時(shí)其物理數(shù)值通常不能劇烈跳變，變化率會(huì)被限制在某范圍內(nèi)，因此速度約束在工業(yè)場景中普遍存在。

式中：xn，xm為兩個(gè)點(diǎn)的數(shù)值；tn，tm為數(shù)據(jù)點(diǎn)對(duì)應(yīng)的時(shí)間戳；smin，smax分別為兩點(diǎn)速度約束的下限和上限。

2.1 基于速度約束的清洗方法

現(xiàn)有的速度約束清洗方法[9，13］，使用了約束模型的中位數(shù)近似求解方法?？疾靾D2a時(shí)間序列X中在時(shí)間窗口w內(nèi)的兩個(gè)點(diǎn)xi和xk，其中xk為待修復(fù)的異常點(diǎn)。

根據(jù)速度約束條件，xk應(yīng)當(dāng)在xi的約束s所限制的范圍內(nèi)，圖2a中短線表示的范圍邊界是數(shù)據(jù)點(diǎn)xk在前時(shí)刻數(shù)據(jù)點(diǎn)xi約束下的最大取值和最小取值，最大取值和最小取值構(gòu)成的邊界限制了xk的取值范圍。

再考查圖2b 中待清洗的數(shù)據(jù)點(diǎn)xk，以xk為起點(diǎn)的窗口w內(nèi)有m個(gè)數(shù)據(jù)點(diǎn)。若已知k+1 時(shí)刻，k+2時(shí)刻…k+m時(shí)刻的數(shù)據(jù)點(diǎn)，則可以依據(jù)速度約束回溯xk的值。圖2b中星號(hào)標(biāo)記表示了xk后面m個(gè)的數(shù)據(jù)點(diǎn)回溯xk范圍的上限xkmin和下限xkmax。

將xk本身和xk前時(shí)刻點(diǎn)約束范圍端點(diǎn)以及后時(shí)刻點(diǎn)xk+1，xk+2，…，xk+m的回溯結(jié)果成候選結(jié)果集，記作Xk，則

為了滿足最小修復(fù)距離原則，位于候選集Xk中間的數(shù)值通常具有較小的修復(fù)距離。把候選結(jié)果集Xk的中位數(shù)記作xmidk，xmidk就作為xk數(shù)據(jù)點(diǎn)的修復(fù)結(jié)果，圖2b中點(diǎn)xk即為修復(fù)結(jié)果，其滿足速度約束。

2.2 周期性數(shù)據(jù)速度約束清洗算法

通常情況下，該方法能完成大多數(shù)清洗任務(wù)[13］，但在部分工業(yè)環(huán)境中，機(jī)械裝備或運(yùn)動(dòng)構(gòu)件往往具有周期運(yùn)轉(zhuǎn)特性，采集的數(shù)據(jù)也呈現(xiàn)周期變化；往復(fù)運(yùn)動(dòng)的組件，其數(shù)值會(huì)有周期突變，例如圖3a展示的盾構(gòu)機(jī)油缸行程數(shù)據(jù)，在一個(gè)掘進(jìn)周期中，油缸以設(shè)定的速度推進(jìn)，完成一環(huán)任務(wù)時(shí)，油缸迅速返回，并為下一環(huán)的推進(jìn)工作做好準(zhǔn)備。

圖3 盾構(gòu)機(jī)樣本數(shù)據(jù)和清洗結(jié)果Fig.3 Sample data of a tunnel boring machine and the cleaning results

而速度約束方法用于此類設(shè)備的數(shù)據(jù)清洗時(shí)，不能獲得良好的清洗結(jié)果。速度約束方法會(huì)把每個(gè)工作周期的數(shù)據(jù)按照連續(xù)數(shù)據(jù)清洗，導(dǎo)致兩個(gè)周期間、周期末尾的數(shù)據(jù)點(diǎn)發(fā)生較大失真，見圖3b。這是由于當(dāng)清洗窗口包含兩個(gè)周期的數(shù)據(jù)時(shí)，每個(gè)周期最后時(shí)刻的突變點(diǎn)會(huì)被視作窗口內(nèi)的異常值，速度約束算法會(huì)將其修改為 “合理值” ，而實(shí)際上是油缸的回退時(shí)造成了該數(shù)據(jù)的突變，這些數(shù)據(jù)應(yīng)被作為真實(shí)數(shù)據(jù)而不能被清洗和修復(fù)。

為提高速度約束算法的適用性，對(duì)周期性數(shù)據(jù)清洗前，應(yīng)先識(shí)別并劃分?jǐn)?shù)據(jù)的周期，限制窗口范圍。以油缸行程數(shù)據(jù)為例，應(yīng)僅對(duì)盾構(gòu)推進(jìn)時(shí)的數(shù)據(jù)進(jìn)行清洗，油缸回退時(shí)的數(shù)據(jù)不應(yīng)當(dāng)參與速度約束求解。

算法1描述了本清洗系統(tǒng)中周期性數(shù)據(jù)速度約束清洗算法（Periodic Data Cleaning Based on Speed Method）的流程。數(shù)據(jù)清洗算法主要面向時(shí)序數(shù)據(jù)中的單點(diǎn)異常值，異常值的識(shí)別依據(jù)式（1），不滿足式（1）約束的被視作為異常值。清洗算法按照最小修復(fù)原則[6］，使得清洗后的序列X’內(nèi)的所有點(diǎn)能夠滿足約束，并且清洗后的值與原始數(shù)據(jù)值盡可能接近。即在滿足速度約束s條件下，為周期性時(shí)間序列X尋找修復(fù)距離最小的修復(fù)序列X’，其中修復(fù)距離指原始時(shí)間序列X中的點(diǎn)x和修復(fù)后的數(shù)據(jù)點(diǎn)x’的差別，記作Δx，如圖2a所示。

算法模型以Kafka Consumer身份接入系統(tǒng)并作為清洗組件。清洗組件先依據(jù)Kafka 配置連接到Kafka 集群，再通過Kafka 主題訂閱待清洗數(shù)據(jù)。清洗組件每隔一段間隔調(diào)用數(shù)據(jù)拉取函數(shù)從集群中獲取一批數(shù)據(jù)，記作DatasRecords。

隨后，清洗組件調(diào)用預(yù)處理函數(shù)對(duì)每條數(shù)據(jù)Record 做格式轉(zhuǎn)換等預(yù)處理，預(yù)處理方式按照不同數(shù)據(jù)格式定制，預(yù)處理后的數(shù)據(jù)被暫存到RangofResult隊(duì)列中等待清洗；當(dāng)RangofResult中的數(shù)據(jù)達(dá)到一定窗口時(shí)進(jìn)行數(shù)據(jù)清洗工作。數(shù)據(jù)清洗工作通過調(diào)用清洗周期時(shí)間序列函數(shù)完成，清洗后的結(jié)果被記錄在CleanedDatas集合中。

算法1中的清洗周期時(shí)間序列函數(shù)會(huì)先用調(diào)用判斷函數(shù)對(duì)隊(duì)列RangOfResult的數(shù)據(jù)逐條判斷，判斷數(shù)據(jù)是否屬于同一周期。周期判斷的方法應(yīng)依據(jù)現(xiàn)場數(shù)據(jù)進(jìn)行特性化設(shè)計(jì)。本例中，有效工作周期和非工作周期數(shù)據(jù)間有明顯的突變，對(duì)圖3a所示的油缸行程數(shù)據(jù)可使用閾值判斷法，大于特定值的數(shù)據(jù)屬于同一周期。當(dāng)檢測到周期突變數(shù)據(jù)時(shí)，終止檢測，并將終止前的一個(gè)切片數(shù)據(jù)記錄在OneSplitPeriod隊(duì)列中。本例還可通過數(shù)據(jù)跳變檢測，數(shù)據(jù)突變檢測等方式對(duì)數(shù)據(jù)進(jìn)行周期性劃分。對(duì)于周期性明確的數(shù)據(jù)，可通過限制清洗窗口來劃分清洗周期。

數(shù)據(jù)被劃分為周期片段后，調(diào)用算法2 清洗周期片段函數(shù)清洗每一個(gè)切片數(shù)據(jù)。最后，清洗組件將算法2的返回結(jié)果寫入存儲(chǔ)系統(tǒng)。

算法1數(shù)據(jù)清洗組件

Algorithm1 DataCleaningModule

輸入 待處理的周期性時(shí)間序列

輸出 清洗后的時(shí)間序列

主函數(shù) 數(shù)據(jù)清洗｛

數(shù)據(jù)清洗消費(fèi)者 = 新的Kafka Consumer（Kafka配置）；

數(shù)據(jù)清洗消費(fèi)者. 連接和訂閱（待清洗數(shù)據(jù)主題）；

循環(huán)｛

DatasRecords = 數(shù)據(jù)清洗消費(fèi)者. 從Kafka拉取數(shù)據(jù)（一段間隔）；

對(duì)于DatasRecords中的每一個(gè)Record｛

如果Record不為空｛

Result = 預(yù)處理（Record）；

RangofResult. 添加（Result）；｝｝

如果RangofResult. 大小＞ 窗口大?。?/p>

CleanedDatas = 清洗周期時(shí)間序列（RangofResult）；

清空RangofResult；｝

寫入數(shù)據(jù)存儲(chǔ)組件（CleanedDatas ）；｝｝

函數(shù) 清洗周期時(shí)間序列（RangofResult）｛

對(duì)于RangofResult中的每一個(gè)Record｛

歸屬的周期 = 判斷（Record ）；

OneSplitPeriod. 添加（歸屬的周期， Record ）；｝

如果（OneSplitPeriod 不為空）｛

清洗后的序列片段 = 清洗周期片段（OneSplitPeriod）；

返回 清洗后的序列片段；｝｝

單個(gè)周期片段數(shù)據(jù)的速度約束清洗流程如算法2 所示[9］，先依據(jù)xk的前時(shí)刻數(shù)據(jù)點(diǎn)xk－1，計(jì)算前時(shí)刻點(diǎn)的約束集再依據(jù)xk后時(shí)刻m個(gè)點(diǎn)xk+1，xk+2，…，計(jì)算xk的后時(shí)刻回溯集。由前時(shí)刻點(diǎn)的約束集和后時(shí)刻點(diǎn)的回溯集組成候選集Xk，從候選集中求出中值xmkid。最后再判斷中值與的包含關(guān)系選出xk的數(shù)據(jù)清洗結(jié)果。候選值xmkid與約束范圍只能有以下幾種關(guān)系：屬于關(guān)系則清洗結(jié)果為xmkid；超過上限清洗結(jié)果為；低于下限清洗結(jié)果為

算法2清洗周期片段

Algorithm2 CleanAFragmentofTimeSeries

輸入 一個(gè)時(shí)間序列片段；

輸出 清洗后的時(shí)間序列片段；

清洗后的時(shí)間序列片段 = 新建 一個(gè)列表；

對(duì)于 一個(gè)時(shí)間序列片段 中的每一個(gè)xk｛

候選集Xk= 新建 一個(gè)列表；

m個(gè)后續(xù)點(diǎn) = 獲取xk后的m個(gè)數(shù)據(jù)點(diǎn)；

對(duì)于m個(gè)后續(xù)點(diǎn) 中的每一個(gè)xnkex｛t

xk的清洗結(jié)果 = 判斷包含關(guān)系（候選集Xk的中位數(shù)，xmaxk，xmink）；

清洗后的時(shí)間序列片段. 添加（xk的清洗結(jié)果）；}

返回 清洗后的時(shí)間序列片段。

3 實(shí)驗(yàn)評(píng)估

為了測試周期性數(shù)據(jù)清洗算法用于周期時(shí)序數(shù)據(jù)的有效性，試驗(yàn)選取了來源于上海申通盾構(gòu)集團(tuán)的中鐵CREC929-932 系列盾構(gòu)機(jī)掘進(jìn)數(shù)據(jù)集進(jìn)行模擬實(shí)驗(yàn)。原始樣本集包含油缸行程、油缸推力等200維數(shù)據(jù)，選取周期性明顯的A組油缸行程數(shù)據(jù)為清洗對(duì)象，再從此列數(shù)據(jù)中選取2 500個(gè)時(shí)序點(diǎn)組成原始序列，如圖4所示。該組數(shù)據(jù)記錄了盾構(gòu)機(jī)的8環(huán)掘進(jìn)過程，圖4 中8 個(gè)上升階段為每個(gè)推進(jìn)過程，圖4中的迅速下降階段為油缸的回退過程。

圖4 油缸行程數(shù)據(jù)和異常值點(diǎn)示例Fig.4 Cylinder stroke data and outliers

所有實(shí)驗(yàn)程序用Java 實(shí)現(xiàn)，其中數(shù)據(jù)生成程序會(huì)模擬工業(yè)現(xiàn)場底層采集單元的數(shù)據(jù)上傳。系統(tǒng)中各組件和算法程序運(yùn)行在Intel（R） Xeon（R） CPU E5-2603 v4 @ 1.70GHz 的CPU 和8GB 內(nèi) 存 的PC上，操作系統(tǒng)為CentOS Linux Release 7.8。5 臺(tái)上述規(guī)格的PC 組成實(shí)驗(yàn)集群，編號(hào)為PC1～PC5，集群配置如表1 所示，其中Zookeeper 用于輔助Kafka集群管理元數(shù)據(jù)信息。

表1 集群配置信息Tab.1 Cluster configuration information

實(shí)驗(yàn)選用移動(dòng)平均法SMA、加權(quán)平均法WMA、速度約束法[9，13］、隨機(jī)森林、神經(jīng)網(wǎng)絡(luò)與周期性數(shù)據(jù)速度約束清洗算法（簡記為周期性清洗算法）對(duì)比。其中機(jī)器學(xué)習(xí)模型對(duì)照組以待清洗的A組油缸行程為標(biāo)簽集建模。模型評(píng)價(jià)指標(biāo)包括：異常數(shù)據(jù)修復(fù)距離和，修復(fù)結(jié)果偏差和，清洗準(zhǔn)確度，指標(biāo)含義見3.2節(jié)。

本實(shí)驗(yàn)中，因盾構(gòu)推進(jìn)中途油缸不可回縮，油缸行程的速度約束值下限選取為0。速度約束的上限依據(jù)施工組織方案中的施工計(jì)劃設(shè)定，油缸推進(jìn)速度最大為80mm·min-1。對(duì)于無特定限制的其他系統(tǒng)可以采用統(tǒng)計(jì)方法選取合理的速度約束值。

實(shí)驗(yàn)使用異常點(diǎn)替換原始序列中的部分樣本點(diǎn)，來模擬工業(yè)現(xiàn)場采集到異常數(shù)據(jù)的場景，驗(yàn)證系統(tǒng)的可用性和算法的有效性。異常點(diǎn)的位置會(huì)對(duì)數(shù)據(jù)的清洗效果產(chǎn)生影響，清洗系統(tǒng)面向周期性數(shù)據(jù)，系統(tǒng)會(huì)檢測并劃分一組數(shù)據(jù)的不同周期，本實(shí)驗(yàn)中異常數(shù)據(jù)只被添加在油缸每個(gè)工作周期的推進(jìn)行程中。

為體現(xiàn)算法在不同異常值分布下的泛化性，實(shí)驗(yàn)使用均勻分布和偏態(tài)分布模擬工程裝備運(yùn)轉(zhuǎn)時(shí)出現(xiàn)的異常值。均勻分布用于模擬裝備在正常工作階段時(shí)普遍出現(xiàn)的異常值。異常數(shù)據(jù)的位置服從均勻分布，樣本集中每個(gè)數(shù)據(jù)點(diǎn)等可能地出現(xiàn)異常；手動(dòng)添加的均勻分布異常點(diǎn)與原本正常點(diǎn)數(shù)值上存在20%～30%的偏差。偏態(tài)分布用于模擬裝備啟動(dòng)階段的高概率異常。周期運(yùn)轉(zhuǎn)的大型裝備，在工作循環(huán)的啟動(dòng)階段，由于環(huán)境復(fù)雜，交叉作業(yè)多，易發(fā)生信號(hào)干擾，初始階段異常值的出現(xiàn)概率高，正常工作階段異常概率相對(duì)較低。在該場景下，實(shí)驗(yàn)使用Zipf分布模擬這種偏態(tài)分布；手動(dòng)添加的偏態(tài)分布異常點(diǎn)與原本正常點(diǎn)數(shù)值上存在20%～30%的偏差。

3.1 清洗算法效果直觀對(duì)比

首先直觀地對(duì)比在兩種分布下使用不同方法的清洗效果，分別在原始序列中添加1%的服從均勻分布的異常點(diǎn)和5%的服從偏態(tài)分布的異常點(diǎn)，圖中圓點(diǎn)為手動(dòng)添加的異常點(diǎn)，如圖4所示。

清洗系統(tǒng)運(yùn)用不同算法清洗序列的結(jié)果如圖5和圖6所示。移動(dòng)平均算法會(huì)將每個(gè)油缸推進(jìn)周期末尾的突變數(shù)據(jù)及其臨近點(diǎn)做平滑處理，影響了部分正常數(shù)據(jù)，如圖5a、5b和圖6a、6b所示。圖5c和圖6c為周期性數(shù)據(jù)清洗算法結(jié)果，算法通過數(shù)據(jù)切片，識(shí)別有效工作階段的數(shù)據(jù)，在保留原始數(shù)據(jù)突變趨勢的條件下較好地完成異常值的清洗。普通速度約束算法未對(duì)數(shù)據(jù)的周期性進(jìn)行識(shí)別，當(dāng)清洗窗口跨越不同周期時(shí)，窗口內(nèi)周期末尾的突變會(huì)被判斷為異常值，并受到速度約束（速度大于0）的限制，被速度約束方法修復(fù)，導(dǎo)致清洗結(jié)果偏離，見圖5d 和圖6d。圖5e、5f和圖6e、6f展示了兩種機(jī)器學(xué)習(xí)方法的清洗效果，機(jī)器學(xué)習(xí)方法能較準(zhǔn)確地識(shí)別異常值，但對(duì)部分異常值的修復(fù)結(jié)果偏差大。

圖5 系統(tǒng)應(yīng)用不同清洗算法的直觀清洗結(jié)果（含服從均勻分布的1%異常值）Fig.5 Cleaning results of different algorithms applied to the system (Including 1 % outliers obeying uniform distribution)

圖6 系統(tǒng)應(yīng)用不同清洗算法的直觀清洗結(jié)果（含服從偏態(tài)分布的5%異常值）Fig.6 Cleaning results of different algorithms applied to the system (Including 5 % outliers obeying skewed distribution)

3.2 數(shù)據(jù)清洗結(jié)果指標(biāo)對(duì)比

為了測試周期性數(shù)據(jù)清洗算法在不同異常數(shù)據(jù)占比時(shí)的清洗效果，對(duì)比各算法在異常點(diǎn)比例為1%、2%、5%…25%的8組樣本條件下的異常數(shù)據(jù)修復(fù)距離和，修復(fù)結(jié)果偏差和，清洗準(zhǔn)確度3種指標(biāo)。

異常數(shù)據(jù)修復(fù)距離是指修復(fù)前數(shù)據(jù)與修復(fù)后數(shù)據(jù)的距離；修復(fù)距離越小，則系統(tǒng)越能反應(yīng)數(shù)據(jù)原有的變化趨勢，修復(fù)距離和是指所有點(diǎn)的修復(fù)距離加和，如圖7所示。修復(fù)結(jié)果偏差是指修復(fù)后的數(shù)據(jù)點(diǎn)和真實(shí)數(shù)據(jù)點(diǎn)的距離，用于判斷系統(tǒng)修復(fù)后的點(diǎn)和真實(shí)數(shù)據(jù)的偏差，偏差和距離越小，則系統(tǒng)清洗后的結(jié)果越能反應(yīng)真實(shí)數(shù)據(jù)。清洗算法的準(zhǔn)確度用被改動(dòng)的數(shù)據(jù)比例表示，比例越少則說明清洗系統(tǒng)的修復(fù)結(jié)果對(duì)原本正常數(shù)據(jù)的影響越小，其清洗準(zhǔn)確程度越高。

圖7 修復(fù)距離和、結(jié)果偏差和示意[9］Fig.7 Repair distance and result distance

清洗系統(tǒng)使用不同算法的修復(fù)距離和如圖8a和圖8b 所示。其中修復(fù)距離隨著異常點(diǎn)的增多而增加，但周期性清洗算法比平滑算法SMA，WMA 具有更小的修復(fù)距離，修復(fù)距離約為平滑算法的1/3或更低，能夠更好的反應(yīng)數(shù)據(jù)原有的變化趨勢。普通速度約束算法的清洗結(jié)果由于偏差較大沒有在圖中展示。機(jī)器學(xué)習(xí)方法受異常值點(diǎn)分布位置的影響，當(dāng)異常值服從均勻分布時(shí)，機(jī)器學(xué)習(xí)方法清洗含10%異常樣本的修復(fù)結(jié)果高于含15%異常樣本的結(jié)果，這是由于部分異常點(diǎn)位于周期突變處，產(chǎn)生了大偏差；當(dāng)服從偏態(tài)分布時(shí)，異常值多數(shù)分布在前半周期，機(jī)器學(xué)習(xí)方法在異常值較少時(shí)具有較好的修復(fù)距離，當(dāng)錯(cuò)誤數(shù)據(jù)增加時(shí)修復(fù)距離明顯增加。在兩種異常值分布條件下，周期性清洗算法均比其他算法具有較小的修復(fù)距離。

圖8 清洗系統(tǒng)應(yīng)用不同算法的修復(fù)質(zhì)量指標(biāo)對(duì)比Fig.8 Comparison of repair quality indicators for different algorithms used in the cleaning system

所有被修復(fù)的數(shù)據(jù)點(diǎn)和真實(shí)數(shù)據(jù)點(diǎn)的偏差值如圖8c和圖8d所示，結(jié)果偏差隨著異常點(diǎn)比例的增加而增加。運(yùn)用平滑類方法時(shí)，周期末尾的突變數(shù)據(jù)放大了平滑方法的修復(fù)偏差，異常點(diǎn)的臨近點(diǎn)會(huì)被清洗，增加了結(jié)果偏差。機(jī)器學(xué)習(xí)方法的結(jié)果偏差和受異常值分布影響大，不同異常分布時(shí)修復(fù)結(jié)果波動(dòng)大。周期性清洗算法其偏差和約為平滑算法的1/3或更低，證實(shí)了通過周期切片的優(yōu)化可較大程度上避免對(duì)正常數(shù)據(jù)的影響。周期性清洗算法在不同異常值分布下均具有較小的修復(fù)結(jié)果偏差，具有更好的泛化性。

不同算法導(dǎo)致數(shù)據(jù)變化的比例如圖8e和圖8f所示。平滑方法會(huì)將異常點(diǎn)和突變點(diǎn)的臨近點(diǎn)清洗，導(dǎo)致了數(shù)據(jù)清洗的準(zhǔn)確度降低；但平滑方法使用均值逐個(gè)修復(fù)數(shù)據(jù)，在不同異常比例時(shí)均具有穩(wěn)定的準(zhǔn)確率，平滑方法對(duì)數(shù)據(jù)集的影響程度約23%。普通的速度約束方法，會(huì)修復(fù)每個(gè)窗口內(nèi)未滿足約束的值，導(dǎo)致周期突變的原始數(shù)據(jù)被大量修改，對(duì)數(shù)據(jù)集樣本點(diǎn)的影響率80%。機(jī)器學(xué)習(xí)方法能較為準(zhǔn)確的修復(fù)異常數(shù)據(jù)，但修復(fù)距離偏大。周期性數(shù)據(jù)速度約束算法克服了普通速度方法對(duì)突變值的缺陷，能較為準(zhǔn)確的對(duì)錯(cuò)誤數(shù)據(jù)進(jìn)行清洗，當(dāng)錯(cuò)誤率增加時(shí)仍對(duì)樣本數(shù)據(jù)影響較小。

綜上，周期性數(shù)據(jù)清洗算法在用于周期性的工業(yè)數(shù)據(jù)修復(fù)時(shí)，且具有良好的泛化性，能用于不同類型的異常值分布。周期性數(shù)據(jù)清洗算法比平滑方法和普通速度約束方法有更小的修復(fù)距離，且修復(fù)結(jié)果能更好的反應(yīng)數(shù)據(jù)變化趨勢。通過周期識(shí)別與切分，提高了周期性清洗算法的準(zhǔn)確度，對(duì)原本正確的數(shù)據(jù)影響較小。機(jī)器學(xué)習(xí)方法在異常值較少時(shí)清洗效果好，隨著異常值比例增多修復(fù)效果波動(dòng)大偏差增加，且受異常數(shù)據(jù)分布影響，而周期性清洗算法在不同數(shù)據(jù)分布下均有良好的清洗效果。

4 結(jié)語

本文搭建了一套面向工業(yè)時(shí)序數(shù)據(jù)的流式清洗系統(tǒng)，系統(tǒng)以Mosquitto作為底層采集單元的匯集中心，通過Flume 連接Mosquitto 和Kafka 實(shí)現(xiàn)整套系統(tǒng)的數(shù)據(jù)流轉(zhuǎn)，將清洗后的數(shù)據(jù)進(jìn)行存儲(chǔ)或進(jìn)分析。該系統(tǒng)相比傳統(tǒng)的離線時(shí)序數(shù)據(jù)清洗系統(tǒng)，具有高吞吐，緩沖，解耦等優(yōu)勢。

本文還設(shè)計(jì)了一種基于速度約束的周期性數(shù)據(jù)清洗方法，在速度約束算法基礎(chǔ)上將周期性數(shù)據(jù)進(jìn)行周期識(shí)別和數(shù)據(jù)分片，解決速度約束算法用于周期突變數(shù)據(jù)的清洗失效問題，更好地完成具有周期性的工業(yè)數(shù)據(jù)清洗任務(wù)。最后通過實(shí)驗(yàn)驗(yàn)證了本系統(tǒng)在處理周期性工業(yè)數(shù)據(jù)時(shí)的可用性，驗(yàn)證了周期性清洗方法比原有速度約束、平滑算法有更小的修復(fù)距離和結(jié)果偏差，驗(yàn)證了周期性數(shù)據(jù)清洗算法對(duì)原始數(shù)據(jù)的影響比例更小，修復(fù)結(jié)果能更好的反應(yīng)真實(shí)數(shù)據(jù)和現(xiàn)有數(shù)據(jù)的變化趨勢；在不同異常數(shù)據(jù)分布和占比條件下，周期性數(shù)據(jù)清洗方法仍具有較好的泛化性。

作者貢獻(xiàn)聲明：

王 耀：進(jìn)行系統(tǒng)構(gòu)建和算法設(shè)計(jì)，論文撰寫；

趙 炯：擬定研究方向，項(xiàng)目規(guī)劃和指導(dǎo)，論文修改和定稿；

周奇才：指導(dǎo)系統(tǒng)構(gòu)建，指導(dǎo)論文修改；

熊肖磊：幫助數(shù)據(jù)分析；

陳傳林、張恒：提供樣本數(shù)據(jù)，分析實(shí)驗(yàn)場景，協(xié)助數(shù)據(jù)分析。