基于Spark和小波分析的水上交通異常數(shù)據(jù)實時檢測方法研究

楊 帆， 何正偉， 劉力榮

(1.武漢理工大學(xué) 航運學(xué)院， 湖北 武漢 430063；2.內(nèi)河航運技術(shù)湖北省重點實驗室， 湖北 武漢 430063；3.國家水運安全工程技術(shù)研究中心， 湖北 武漢 430063)

隨著“一帶一路”倡議的穩(wěn)步推進，水上交通已經(jīng)成為中國同各國之間貿(mào)易往來的重要方式。水路運輸由于其成本低、可運載貨物量大等優(yōu)勢而成為重要的運輸方式，水上貿(mào)易的推進能夠極大促進國家之間經(jīng)濟貿(mào)易的往來與發(fā)展。然而，隨著水上交通越發(fā)頻繁，運輸量逐年增加，水路運輸所帶來的交通事故也在不斷增長。船舶自動識別系統(tǒng)(Automatic Identification System，AIS)是目前監(jiān)測與分析船舶狀況、維持水上交通穩(wěn)定進行的重要工具。通過分析AIS設(shè)備中的AIS數(shù)據(jù)，能夠?qū)λ辖煌ǖ慕煌髁?、交通擁堵狀況、交通穩(wěn)定性等方面進行判別與預(yù)測。然而，隨著水路運輸?shù)脑鲩L，AIS數(shù)據(jù)量也在不斷增加，傳統(tǒng)的數(shù)據(jù)處理方式很難滿足如此大規(guī)模的數(shù)據(jù)處理要求。因此，需要通過大數(shù)據(jù)處理平臺對AIS數(shù)據(jù)進行實時、有效地分析和處理。其中，AIS數(shù)據(jù)中存在大量的異常交通信息，這些異常信息往往包含著錯誤的船舶狀態(tài)與交通信息，因此，能夠?qū)崟r、高效地對異常信息進行檢測十分重要。

為了檢測異常的交通數(shù)據(jù)，不少研究者做了充分的調(diào)研與研究。Turochy[1]通過交通流參數(shù)的閾值控制對交通異常數(shù)據(jù)進行檢測；陳德旺等[2]提出3種判斷快速路交通流異常數(shù)據(jù)的算法，并討論了這3種算法之間的集成；李成兵等[3]提出一種基于最小二乘支持向量機的交通異常數(shù)據(jù)檢測方法；Chen Shu-yan等[4]通過對離線數(shù)據(jù)進行數(shù)據(jù)挖掘從而檢測出交通異常數(shù)據(jù)。通過對之前的研究方法進行分析，可以看出目前的研究主要是通過對離線數(shù)據(jù)進行分析和檢測，很少涉及到對數(shù)據(jù)的在線實時分析和檢測。本設(shè)計通過Spark大數(shù)據(jù)處理平臺對AIS數(shù)據(jù)進行分析和處理，借助小波分析的方法對船舶交通流量、船舶速度與船舶狀態(tài)進行分析，檢測出異常的交通數(shù)據(jù)，從而減小異常數(shù)據(jù)對水上交通研究的干擾。

1 異常數(shù)據(jù)檢測方法

1.1 水上交通異常數(shù)據(jù)

目前船舶AIS數(shù)據(jù)中包含了船舶運行的大量數(shù)據(jù)，是分析水上交通數(shù)據(jù)的重要工具。水上交通的異常數(shù)據(jù)主要包含以下幾類：a)典型錯誤數(shù)據(jù)，這些數(shù)據(jù)主要是一些明顯的交通異常數(shù)據(jù)，比如緯度91°，船舶速度100等錯誤數(shù)據(jù)；b)異常交通情況數(shù)據(jù)，這些數(shù)據(jù)包含了異常的交通狀況，相比前后時間點的交通情況數(shù)據(jù)來說，這些數(shù)據(jù)有著明顯的交通信息突變，因此被判斷為異常數(shù)據(jù)；c)缺失數(shù)據(jù)，這部分交通數(shù)據(jù)為空，沒有包含任何的船舶運行數(shù)據(jù)。本設(shè)計主要是對異常交通情況數(shù)據(jù)進行分析與檢測，提高交通數(shù)據(jù)的可靠性。

1.2 AIS數(shù)據(jù)接收

由于AIS數(shù)據(jù)量龐大，包含船舶交通信息量多，傳統(tǒng)的單機處理模式已經(jīng)無法滿足大規(guī)模的AIS數(shù)據(jù)處理的要求。本設(shè)計通過Spark數(shù)據(jù)處理引擎來對AIS數(shù)據(jù)進行分析與處理。Spark是繼Hadoop大數(shù)據(jù)處理平臺之后的又一大數(shù)據(jù)處理工具，它在繼承Hadoop中MapReduce計算框架的同時，又有著出色的并行計算能力，能夠快捷、高效地對龐大的數(shù)據(jù)集進行處理[5-6]。

由于AIS數(shù)據(jù)通過基站向本地進行傳輸，因此，數(shù)據(jù)傳輸基本能夠保證實時性。在數(shù)據(jù)接收階段，采用了異步回調(diào)機制對數(shù)據(jù)進行認證和接收，首先向服務(wù)器發(fā)送帶有密碼的套接字，服務(wù)器進行接收并且校驗，如果密碼正確，則進行AIS數(shù)據(jù)的傳輸，從而保證系統(tǒng)的安全性。Spark將AIS數(shù)據(jù)轉(zhuǎn)換成彈性分布式數(shù)據(jù)集(Resilient Distributed Datasets，RDD)，通過自身的算子對數(shù)據(jù)進行一系列轉(zhuǎn)換，最終的處理結(jié)果將通過分布式文件系統(tǒng)進行存儲。圖1為AIS數(shù)據(jù)接收部分的流程圖。

圖1 AIS數(shù)據(jù)接收

1.3 異常數(shù)據(jù)檢測

本設(shè)計通過小波分析對水上交通的異常數(shù)據(jù)進行分析和檢測。相比傅里葉分析，小波分析通過對信號進行一系列的伸縮、平移等轉(zhuǎn)換，對信號進行多角度、多分辨的處理，從而解決傅里葉變換無法解決的問題，使得小波分析方法有著更高的應(yīng)用價值。

在對水上交通數(shù)據(jù)進行處理時，假設(shè)處理的信號為f(t)∈L2(R)，其中，L2(R)為所有平方可積函數(shù)所組成的空間。因此，通過小波函數(shù)進行小波變換后得到：

(1)

(2)

其中，a為膨脹參數(shù)，b為平移參數(shù)，此時式(1)稱為離散小波變換。當對AIS交通數(shù)據(jù)進行多層小波變換時，假設(shè)此時輸入信號為s，通過m層變換后得到子空間為(d1，d2，…，dm，am)，并且滿足以下關(guān)系：

s=d1⊕a1=d1⊕d2⊕a2=d1⊕d2⊕…⊕dm⊕am，

(3)

其中，⊕表示異或邏輯運算，di為信號s的高頻部分，ai為信號s的低頻部分。因此，通過小波變換將AIS中的數(shù)據(jù)進行多層分解，能夠?qū)Ω哳l和低頻部分進行分別處理[7-9]。

在AIS數(shù)據(jù)中，由于信號傳輸過程受到干擾，或者信號中夾雜著噪聲等因素導(dǎo)致交通數(shù)據(jù)出現(xiàn)異常，這些異常數(shù)據(jù)一般存在于信號分解后的高頻部分。因此，異常檢測過程就是通過小波分析的方法，對高頻數(shù)據(jù)進行處理，處理后的高頻數(shù)據(jù)和低頻數(shù)據(jù)進行重構(gòu)，從而得到重構(gòu)后數(shù)據(jù)s′，通過對s′和s進行分析，最終判斷異常數(shù)據(jù)出現(xiàn)的位置以及異常數(shù)據(jù)的個數(shù)，并且對異常數(shù)據(jù)根據(jù)不同情況采取數(shù)據(jù)清除或數(shù)據(jù)矯正，從而完成異常數(shù)據(jù)的檢測功能。

1.4 整體功能設(shè)計

本設(shè)計借助大數(shù)據(jù)處理平臺對AIS數(shù)據(jù)進行處理，通過小波分析的方法將數(shù)據(jù)進行多層分解，通過對高頻數(shù)據(jù)部分進行處理后，將數(shù)據(jù)進行重構(gòu)。通過對原始數(shù)據(jù)與重構(gòu)數(shù)據(jù)的分析與檢測得到殘差序列，從而判斷出異常數(shù)據(jù)的位置和個數(shù)，最終的處理結(jié)果寫入分布式文件系統(tǒng)HDFS中[10-11]。

圖2 系統(tǒng)總體框架

首先，AIS數(shù)據(jù)經(jīng)過服務(wù)器校驗并發(fā)送后，客戶端開始對數(shù)據(jù)進行接收和讀取，需要對AIS數(shù)據(jù)進行預(yù)處理，包含去掉數(shù)據(jù)為空的數(shù)據(jù)并且對AIS數(shù)據(jù)進行解析；解析后的AIS數(shù)據(jù)會通過Spark大數(shù)據(jù)處理系統(tǒng)進行處理，具體的處理操作包括map、reduceBykey、groupBykey等算子，統(tǒng)計出水上交通選定區(qū)域的交通流量、平均交通速度以及船舶狀態(tài)；之后，AIS數(shù)據(jù)會通過小波分析進行處理，經(jīng)過對AIS數(shù)據(jù)多層分解、高頻信號處理等過程，對數(shù)據(jù)進行重構(gòu)；分別將數(shù)據(jù)和Spark大數(shù)據(jù)處理平臺統(tǒng)計分析后所得結(jié)果進行比對和計算，將異常數(shù)據(jù)檢測出來，并且通過對數(shù)據(jù)進行分析，判斷是否完成對數(shù)據(jù)的修復(fù)或提出，最終對整個異常數(shù)據(jù)檢測系統(tǒng)性能進行分析和評價。整個設(shè)計的總體框架如圖2所示。

2 Spark數(shù)據(jù)處理方法

2.1 AIS大數(shù)據(jù)處理

AIS數(shù)據(jù)要通過Spark大數(shù)據(jù)處理平臺對數(shù)據(jù)進行初步處理和計算。本設(shè)計中，選擇交通情況中的3個指標進行異常數(shù)據(jù)分析：交通流量情況、船舶平均速度和船舶狀態(tài)。其中，交通流量情況通過對AIS數(shù)據(jù)中的動態(tài)數(shù)據(jù)船舶MMSI(特定船舶的編號)和Spark的reduceBykey算子對特定區(qū)域船舶數(shù)量進行統(tǒng)計，從而得到不同時間段的水上交通流量情況。船舶平均速度通過對AIS動態(tài)數(shù)據(jù)中船舶對地速度進行統(tǒng)計，通過Spark的aggregate函數(shù)對速度進行求解平均值，從而計算出不同時間段的船舶平均速度。船舶狀態(tài)通過AIS動態(tài)數(shù)據(jù)中的船舶狀態(tài)進行統(tǒng)計分析，通過統(tǒng)計不同時間段船舶的狀態(tài)得到船舶狀態(tài)的總體情況。通過3個不同的算法來實現(xiàn)數(shù)據(jù)的處理，其中算法1為交通流量情況統(tǒng)計，算法2為船舶平均速度計算，算法3為船舶狀態(tài)統(tǒng)計，具體的算法如下：

算法1 交通流量算法

Input：AIS動態(tài)數(shù)據(jù)

Output：船舶交通流量

1.ais←HDFS

2.aisf←ais.filter(._=0)

3.aisf,s←aisf.split("，").split(" ")

4.aisf,s,m←aisf.map(_.toInt)

5.f←aisf,s,m.reduceBykey

其中，ais表示從HDFS中讀取的原始AIS數(shù)據(jù)，aisf表示去掉缺失數(shù)據(jù)后的的AIS數(shù)據(jù)，aisf，s表示去掉原始數(shù)據(jù)中的“，”和空格的AIS數(shù)據(jù)，aisf，s，m為轉(zhuǎn)換為整數(shù)類型后的AIS數(shù)據(jù)。

算法2 平均速度算法

Input：AIS動態(tài)數(shù)據(jù)

Output：船舶平均速度

1.ais←HDFS

2.aisf←ais.filter(._<20)

3.aisf，s←aisf.split("，").split(" ")

4.aisf，s，m←aisf.map(_.toInt)

算法3 船舶狀態(tài)統(tǒng)計算法

Input：AIS動態(tài)數(shù)據(jù)

Output：各狀態(tài)船舶數(shù)量

1.ais←HDFS

2.aisf←ais.filter(._=0)

3.aisf，s←aisf.split("，").split(" ")

4.aisf，s，m←aisf.map(_.toInt)

5.mumstate←aisf，s，m.reduceBykey

其中，mumstate表示通過reduceBykey計算后的各狀態(tài)下的船舶數(shù)量值。

2.2 小波分析處理

通過小波分析對船舶異常數(shù)據(jù)進行檢測。首先需要讀取HDFS分布式文件系統(tǒng)中的AIS數(shù)據(jù)，并且選取小波函數(shù)和小波分析的層數(shù)，本設(shè)計中選擇的小波函數(shù)為db3小波函數(shù)，分解層數(shù)為3層，通過對數(shù)據(jù)進行多層次分解后，需要將分解后的數(shù)據(jù)的高頻部分進行閾值選擇。通過對船舶交通流量、平均速度和船舶狀態(tài)進行處理后，可以確定3種交通情況的合理閾值大小，對于不滿足閾值大小的異常數(shù)據(jù)進行選取并且輸出，并且通過異常數(shù)據(jù)分析部分對異常數(shù)據(jù)進行判斷。小波分析部分的算法為

算法4 小波分析算法

Input：AIS動態(tài)數(shù)據(jù)

Output：小波分析結(jié)果

1.ais←HDFS

2.aisf←function=db3，layer=3

3.aisf，s←aisf.wlen

4.aisf，s，m←aisf，s.wlen

5.mumstate←aisf，s，m.wlen

其中，ais表示從HDFS中讀取的原始AIS數(shù)據(jù)，aisf表示選擇的小波函數(shù)為db3小波函數(shù)、分解層數(shù)為3層處理后的AIS數(shù)據(jù)，aisf，s、aisf，s，m、mumstate分別表示通過wlen函數(shù)分解處理后的結(jié)果。

2.3 異常數(shù)據(jù)分析

(4)

通過剩余誤差e求解出待檢測數(shù)據(jù)的標準差σ，計算公式為

(5)

如果小波分析中的異常數(shù)據(jù)和標準差之間的關(guān)系為vi≤3σ，此時可以對異常數(shù)據(jù)進行修復(fù)；如果異常數(shù)據(jù)與標準差之間的關(guān)系為vi>3σ，表示異常數(shù)據(jù)與標準差之間的差值過大，此時選擇對異常數(shù)據(jù)進行剔除。該部分的處理算法為

算法5 異常數(shù)據(jù)分析算法

Input：待檢測異常數(shù)據(jù)

Output：異常數(shù)據(jù)分析結(jié)果

1.ais←HDFS

圖3 3種集群環(huán)境下的數(shù)據(jù)處理時間

2.aisf←e，σ

3.If(aisf<=3σ)

aisf，s←aisf

4.If(aisf>3σ)

delete aisf

5.out←aisf，s

3 案例分析

3.1 數(shù)據(jù)來源

本文選取數(shù)據(jù)的地理位置是經(jīng)緯度分布在(東經(jīng)120.725°，北緯32.049°)到(東經(jīng)120.835°，北緯31.981°)之間的一個矩形區(qū)域內(nèi)的航道，時間范圍為2017年2月6日—2月10日。這個范圍的航道區(qū)域是南通航段，在長江航道中位置特殊且十分重要，它緊靠長江入?？冢情L江航道上船舶入海的必經(jīng)航道，也是海上船舶進入長江的必經(jīng)航道。南通航段內(nèi)船舶通航量巨大，船舶水上作業(yè)十分繁忙。

3.2 大數(shù)據(jù)處理結(jié)果

AIS大數(shù)據(jù)處理的集群環(huán)境包括了主節(jié)點Master和3個從節(jié)點Slave1、Slave2、Slave3，分別通過算法1、算法2、算法3對數(shù)據(jù)進行處理，通過Spark自身的算子分布式計算，分別計算得出船舶交通流量、區(qū)域船舶平均速度和船舶狀態(tài)。數(shù)據(jù)處理時間如圖3所示。

從圖中處理時間可以得出，通過reduceBykey算子對船舶交通流量進行統(tǒng)計時，隨著數(shù)據(jù)量的增長，數(shù)據(jù)處理時間也有所增加，此時4節(jié)點的處理性能最好，處理時間最短；通過aggragate函數(shù)對船舶平均速度進行求解時，當數(shù)據(jù)量大于100萬條時，2節(jié)點與4節(jié)點的處理時間基本穩(wěn)定在2000 ms；對不同船舶的狀態(tài)進行統(tǒng)計過程中，多節(jié)點的處理優(yōu)勢也隨著數(shù)據(jù)量的增長而逐步體現(xiàn)，數(shù)據(jù)量達到100萬條時，數(shù)據(jù)處理時間基本穩(wěn)定。

通過算法1、算法2、算法3對數(shù)據(jù)處理的時間可以得出，Spark大數(shù)據(jù)平臺運用分布式集群的處理機制，能夠提高數(shù)據(jù)的處理速度，并且對于大規(guī)模數(shù)據(jù)來說，分布式的處理環(huán)境比單機的處理環(huán)境處理性能更好，處理時間更短，數(shù)據(jù)處理過程更加穩(wěn)定高效。

3.3 異常數(shù)據(jù)檢測結(jié)果

通過小波分析對AIS數(shù)據(jù)進行分解和重構(gòu)后，可以檢測出異常的交通信息，檢測結(jié)果如圖4所示。分別對船舶交通流量信號、船舶平均速度信號和船舶狀態(tài)信號通過小波分析檢測異常數(shù)據(jù)并進行分析后，采用最小二乘的方法，對檢測出的AIS數(shù)據(jù)中的異常信息進行分析，分別記錄檢測出的不同交通數(shù)據(jù)類型中的異常數(shù)據(jù)、檢測中的誤判數(shù)據(jù)和漏判數(shù)據(jù)，并計算出系統(tǒng)對于異常數(shù)據(jù)檢測的準確率。表1為各交通數(shù)據(jù)類型的異常數(shù)據(jù)分析結(jié)果。

圖4 異常數(shù)據(jù)檢測結(jié)果

數(shù)據(jù)類型異常數(shù)據(jù)/條誤判數(shù)據(jù)/條漏判數(shù)據(jù)/條準確率/%船舶流量62332194.6船舶速度54281695.6船舶狀態(tài)71432493.3

通過對異常數(shù)據(jù)進行分析可以得出，檢測出的異常數(shù)據(jù)基本符合AIS數(shù)據(jù)的異常特征，異常數(shù)據(jù)的檢測準確度較高，能夠?qū)IS數(shù)據(jù)的異常信息進行快速、準確的檢測和識別，相比傳統(tǒng)的數(shù)據(jù)檢測方法，本系統(tǒng)在準確性和實時性方面有更大的優(yōu)勢。

4 小 結(jié)

為了對水上交通的異常數(shù)據(jù)進行檢測，本文提出了一種基于大數(shù)據(jù)處理平臺的AIS異常數(shù)據(jù)處理與分析方法，通過對龐大的AIS數(shù)據(jù)集進行在線、實時地處理和分析，找出交通數(shù)據(jù)中的異常數(shù)據(jù)并對異常數(shù)據(jù)進行分析。實驗結(jié)果表示，異常數(shù)據(jù)符合水上交通數(shù)據(jù)的實際情況，并且能夠在短時間內(nèi)快速、穩(wěn)定地完成數(shù)據(jù)的處理和檢測。

由于AIS數(shù)據(jù)量大，包含船舶動態(tài)數(shù)據(jù)和靜態(tài)數(shù)據(jù)，傳統(tǒng)的數(shù)據(jù)處理模式無法滿足處理速度的要求。因此，本文首先構(gòu)建基于Spark的大數(shù)據(jù)處理平臺，通過Spark自身的算子，在集群環(huán)境中對AIS數(shù)據(jù)進行并行化的處理，處理結(jié)果寫入分布式文件系統(tǒng)HDFS中，從而保證了數(shù)據(jù)處理的實時性。

其次，本文對水上交通數(shù)據(jù)中的3個重要指標：水上交通流量、區(qū)域船舶平均速度和船舶狀態(tài)，通過3種算法分別進行處理。采用小波分析的方法將AIS數(shù)據(jù)進行多層分解和重構(gòu)，通過重構(gòu)后的異常信號檢測結(jié)果和殘差序列信號圖找出交通數(shù)據(jù)中的異常數(shù)據(jù)。

最后，通過對異常數(shù)據(jù)進行檢測和分析，確定對異常數(shù)據(jù)是否檢測完成，是否存在誤判、漏判的現(xiàn)象。統(tǒng)計各個交通指標的異常數(shù)據(jù)情況和系統(tǒng)的檢測準確率。

通過大數(shù)據(jù)處理平臺和小波分析相結(jié)合的方法，本文提出的水上交通異常數(shù)據(jù)監(jiān)測分析系統(tǒng)能夠滿足水上交通的實際情況。然而，本文主要是集中在對異常數(shù)據(jù)的檢測和分析，未來的研究中，將會更多地引入對異常數(shù)據(jù)的修正和預(yù)測，并且將通過可視化的工具實時接收處理后的交通數(shù)據(jù)，以更加直觀的方式展示處理結(jié)果，為海事部門與海洋環(huán)境保護機構(gòu)提供更加高效、穩(wěn)定的監(jiān)管服務(wù)。