基于Spark的模糊聚類算法實現(xiàn)及其應用

吳云龍，李玲娟

(南京郵電大學 計算機學院，江蘇 南京 210023)

0 引 言

入侵檢測系統(tǒng)(intrusion detection systems，IDS)[1]在保障企業(yè)網絡的安全性方面扮演著至關重要的角色。為了有效監(jiān)測訪問行為的異常與否，數據挖掘技術被廣泛應用到入侵檢測系統(tǒng)中。數據挖掘算法分為兩類：有監(jiān)督和無監(jiān)督[2]。無監(jiān)督的聚類算法有K-Means、模糊C均值聚類(fuzzy c-means，F(xiàn)CM)等。FCM[3]在傳統(tǒng)C-均值算法中添加隸屬度矩陣來判定某條數據所屬類別。這種算法在運算過程中需要多次迭代，當數據量特別大或者數據樣本屬性很多的時候，運算速度會大大降低。為了提高FCM算法對大數據的聚類效率，設計了FCM算法基于具有實時計算特點的大數據計算平臺Spark[4]的并行化實現(xiàn)方法。該方法采用HDFS分布式存儲底層數據，運用RDD對數據進行轉換，用持久化技術進行中間結果的重用。為了檢驗該并行化FCM算法的有效性，用該算法對KDD CUP 99數據集聚類來進行入侵檢測，首先對入侵檢測數據集KDD CUP 99[5]進行了特征值提取等預處理，然后利用實現(xiàn)的FCM算法對訓練數據集聚類得到聚類中心，再用測試數據集進行聚類效果檢驗。

1 相關技術

1.1 FCM算法分析

聚類分析可以在大量數據中挖掘隱藏的規(guī)律[6]。從聚類的結果來看，聚類算法可以分為兩大類：軟聚類和硬聚類。硬聚類是把某一個數據點明確劃分到某一個類；軟聚類是將樣本個體通過隸屬度標識出與各個類簇的隸屬關系[7]，模糊C均值聚類算法FCM是經典的軟聚類算法[8]。模糊的概念是指，用[0,1]之間的隸屬度來確定某個點屬于某個類的程度。每條數據對所有類中心點的隸屬度之和為1。FCM算法的步驟[9]如下：

步驟1：指定分類數和加權指數m，確定迭代次數，隨機生成初始化隸屬度矩陣并滿足式1的約束條件。

(1)

其中，j表示第j條數據；uij表示第j條數據隸屬于第i個中心點的隸屬度，取值范圍為[0,1]。

步驟2：按聚類中心的迭代公式(見式2)，根據uij構成的隸屬度矩陣U，計算聚類中心。

(2)

其中，m是一個加權指數，可以控制聚類結果的模糊程度，取值范圍為m∈(1,∞)。

步驟3：按式3計算目標函數J。

(3)

其中，dij=‖ci-xj‖，為第j條數據與第i個中心點的歐氏距離。

如果相鄰兩次迭代趨于收斂，小于指定閾值或者超過指定迭代次數，則結束；否則執(zhí)行步驟4。

步驟4：根據新的聚類中心，按隸屬度的迭代公式(式4)重新計算隸屬度矩陣U，并返回到步驟2。

(4)

1.2 Spark平臺

Spark是大規(guī)模數據的通用分布式并行計算平臺。與Hadoop MapReduce編程模型不同，Spark利用RDD(resilient distributed datasets)進行計算，運算過程中的輸出結果可以直接存于內存中。Spark以其分布式內存計算的特點，可以提升需要多次迭代的算法的效率。

RDD是Spark的核心。邏輯上，RDD存放數據；物理上，真實數據存放在各個分區(qū)中，每個分區(qū)可能位于不同的Worker節(jié)點之上。RDD的數據在默認情況下都放在內存中，因而Spark對內存的要求也很高。當內存不能滿足運行需求時，它會把RDD數據序列化到磁盤。RDD具有容錯性，由于把數據同時存放在多個分區(qū)中，當某個節(jié)點出現(xiàn)分區(qū)數據丟失或者損壞等故障時，會自動重建丟失數據。

Spark提供了豐富的算子來操縱RDD，每個算子發(fā)出的指令最終會落實到對每個分區(qū)里數據的操作上。開發(fā)者通過Scala、Java或者Python調用RDD算子操作分布式數據集。RDD根據算子是否立即執(zhí)行，可以分為兩類：Transformation類和Action類。Transformation類操作會發(fā)生延遲計算，不會真正運行，需要Action類操作的觸發(fā)才提交作業(yè)。

運算過程中，會生成多個RDD，從一個RDD轉換到另一個RDD。Action類每次操作都會得到結果，如collect()、first()等。

Spark有4種運行模式：本地模式、Standalone偽分布式模式、基于Mesos的集群模式和基于YARN的集群模式。在Standalone模式下，Spark的架構如圖1所示。

圖1 Spark架構(Standalone模式下)

ClusterManager為Master(主節(jié)點)，控制整個集群，監(jiān)控Worker。Worker為從節(jié)點，負責控制計算節(jié)點，啟動Executor或Driver。RDD的每個分區(qū)可能在不同的Worker機器上，每臺Worker節(jié)點可能有多個Executor，每個Executor可能有多個Task在運行，Task和分區(qū)是一一對應的，所以partition分區(qū)的數量也影響著Task的并發(fā)程度[10]。

Standalone模式下任務提交運行的步驟是：Client提交應用，Master接受Client提交的任務并命令Worker啟動Driver；Driver向Master或者資源管理器申請資源，某種意義上，Driver就是SparkContext，執(zhí)行應用程序中的main()函數，創(chuàng)建SparkContext，SparkContext負責與ClusterManager通信，對Spark進行初始化，并創(chuàng)建DAGScheduler作業(yè)調度模塊和TaskScheduler任務調度模塊等；從Hadoop的HDFS加載數據，生成RDD，構建RDD DAG，將DAG圖分解成Stage(當碰到Action操作時，就會催生Job；每個Job中含有1個或多個Stage，一個Stage包含一個或多個Task，Stage一般在獲取外部數據和Shuffle之前產生)；DAGScheduler把一個Spark作業(yè)轉換成Stage的DAG(directed acyclic graph，有向無環(huán)圖)，根據RDD和Stage之間的關系找出開銷最小的調度方法，然后把Stage以TaskSet的形式提交給TaskScheduler，在解析DAG圖時，根據Shuffle來劃分Job內部的Stage，并給出Stage之間的依賴關系；TaskScheduler借助ActorSystem將任務提交給集群管理器；Task在Executor上運行，運行完畢釋放所有資源。

2 Spark平臺上FCM算法的并行化實現(xiàn)

2.1 算法實現(xiàn)要點

(1)避免反復讀取同樣的數據。

文中采用RDD緩存(cache)，把在計算過程被多次用到的數據cache到內存。

(2)中間結果的持久化。

Spark的Transformations類操作會延遲執(zhí)行，它在Action類算子提交任務到Spark Context時觸發(fā)，如果計算中其他的Action類操作用到中間結果，可以手動調用persist將中間結果做持久化操作。persist可以設置不同參數，不同的參數代表不同的存儲級別。如果把persist參數設置為MEMORY_ONLY，其功能同cache功能。

(3)減少通信開銷。

broadcast是以類似洪泛的方式廣播數據。它是只讀變量，保持數據一致性，可以把迭代過程中的聚類中心廣播到各個分區(qū)，在FCM算法的實現(xiàn)過程中，會不斷迭代產生新的中心點，需要把新的中心點廣播到各個節(jié)點，使得下一次只計算每條數據和新的中心點之間的距離。這樣可以高效地分發(fā)變量，減少通信開銷。

2.2 基于Spark的FCM并行化流程

基于Spark的FCM算法并行化實現(xiàn)流程見圖2。

圖2 算法實現(xiàn)流程

Step1：把文本數據上傳到Hadoop的HDFS文件系統(tǒng)，在應用程序中通過sc.textFile()加載進RDD，然后把數據轉換成稠密向量，這里把數據cache到內存，計算過程中數據直接從內存中獲取。對數據做L2范式，即求向量各元素的平方和然后求平方根，并且做持久化。

Step2：初始化聚類中心，采用takeSample API隨機抽取部分訓練數據集的方法獲得初始聚類中心，并把聚類中心廣播給各個partition。開始計時，便于最后統(tǒng)計運行時間。

Step3：利用mapPartiotions對每個分區(qū)進行計算。計算數據點到各中心的隸屬度和數據點到各簇的距離和，隸屬度滿足式1。距離的計算可以調用Spark MLlib提供的MLUtil.fastsquaredDistance(x1,x2)，x1和x2數據結構為數據點向量和其L2范式的組合；隸屬度的計算依據式2。

Step4：用reduceByKey和collectAsMap得到sums和fuzzycounts。sums表示在并行度為i情況下，數據點到各簇的距離和；fuzzycounts表示在并行度為i情況下，各數據點到類中心的隸屬度和。

Step5：基于sums和fuzzycounts，依據式4計算新的中心點，并比較相鄰兩次中心點有沒有發(fā)生改變，如果未發(fā)生改變則更換標志位，迭代結束，最后返回預測模型FuzzyCMeansModel，它包含了聚類中心和m值。否則迭代次數加1，繼續(xù)迭代。

Step6：用FuzzyCMeansModel的預測函數來預測數據點對某個簇的隸屬度([0,1])。把最終結果存入文本，用于計算聚類準確率。

3 Spark平臺上基于FCM的入侵檢測實驗

3.1 KDD CUP數據集

實驗選取了KDD CUP 99的10%的數據集。這是一個模擬網絡攻擊的數據集，包含了41個網絡屬性，每條記錄結尾的label顯示了正常還是攻擊。按照大類來分，它包含了正常和異常兩種類型的數據。正常訪問的label為Norml，異常數據包括四大攻擊：拒絕服務攻擊DOS、遠程登錄攻擊R2L、非法獲取Root用戶特權攻擊U2R、探測式攻擊Probe。

3.2 數據預處理

在數據挖掘中，從數據源得到的原始數據可能會存在冗余性、空值或者數據不完整等問題。為了提高挖掘的質量和計算速度，往往通過數據清洗、數據降噪、數據降維、數據集成、數據轉換和數據簡約等手段進行預處理[11]。在實驗中，為了提高聚類效果，對數據做了如下預處理。

(1)數據簡約。

有些屬性對聚類效果不僅沒有幫助，還會大大降低數據挖掘的效率，可以采用屬性選擇和數據采樣技術加以簡化。文中采用屬性選擇來去除對聚類效果意義不大的屬性，參考文獻[12]，保留了7個特征：Duration、Protocol_type、Service、Flag、Logged_in、Count、Srv_count。用python腳本提取對應的特征值并保存到另外一份文本中，特征值提取代碼如下：

#需要提取的特征屬性

index=[0,1,2,3,11,22,23]

result=[]

temp=[]

tmp=[]

for line in open("resourcedata"):

tmp=line.split(",")

for i in index:

temp.append(tmp[i])

result.append(temp)

temp=[]

#把result列表數據寫到文件中

(2)數據轉換。

數據轉換就是把數據從一種形式轉換到另外一種形式。把數據集中的字符屬性與數值屬性轉換，進行歸一化，以便于FCM算法的計算。數據用python腳本處理的過程是：首先將保留的7個屬性中的字符型屬性定義為枚舉；然后讀取字符屬性，按照其屬性值轉換為枚舉中定義的值；最后再把結果保存到數據文件中。相應的python偽代碼如下：

fromenum import Enum

classprotocol_type(Enum):

tcp='1'

udp='2'

icmp='3'

protocol_type[tcp].value

(3)空值處理。

空值作為數據對象的值，參與運算仍然是空值。數據挖掘中，空值的存在會使運算混亂，甚至可能無法完成。文中把空值置為“0”。

預處理前數據為41個屬性，其中還有空值；預處理后保留了7個重要屬性，空值也做了處理。

3.3 實驗環(huán)境

實驗中搭建了單機環(huán)境和Standalone集群環(huán)境。單機環(huán)境硬件為HP g6-2025TX，CPU型號為Intel CORE i5-3210M，雙核處理器，內存8 G，硬盤讀寫速度為415 MB/s。軟件環(huán)境為Ubuntu 14.04，JDK版本為JDK1.8，Hadoop版本為2.7.3，Spark版本為2.1，Scala版本為2.10.4。Standalone集群環(huán)境為Master節(jié)點一臺，Worker節(jié)點兩臺，節(jié)點軟件配置與本地模式相同。

3.4 聚類準確度分析

對相同樣本在不同模式下運行FCM算法得到的檢測率如圖3所示。考慮到數據集包含四大類攻擊和正常類型，聚類中心數設置為5。

圖3 在單機和Standalone集群環(huán)境下的檢測率對比

檢測率為：

(5)

由圖3可以看出，F(xiàn)CM算法不會因為在分布式集群環(huán)境下處理數據而影響聚類結果，這說明FCM算法按照文中設計的方法在Spark上并行化運行，具有較好的魯棒性和準確度。

預處理前后的數據上的檢測率對比結果如圖4所示?？梢钥闯?，預處理后的數據上的檢測率更高。

圖4 Standalone集群環(huán)境下數據

3.5 聚類時效對比

預處理前后的數據的聚類時間對比結果如圖5所示。可以看出，對預處理后的數據的處理速度更快。

圖5 Standalone集群環(huán)境下數據預

對預處理后數據分別在單機環(huán)境和Standalone集群環(huán)境下聚類的時耗對比結果如圖6所示。

圖6 單機和集群環(huán)境下預處理后數據的聚類時間對比

可以看出，當數據量不是很大時，單機環(huán)境和集群環(huán)境處理數據消耗的時間相近。集群環(huán)境下很多時間用在Shuffle過程中，甚至單機環(huán)境下處理數據的時效更高。這說明：小數據量并不適合在集群下處理。但是隨著數據量的增多，集群環(huán)境對數據的處理速度明顯快于單機環(huán)境。

4 結束語

文中主要研究FCM算法如何在Spark平臺并行化實現(xiàn)，擴充了Spark MLlib，并將并行化的FCM算法應用于對入侵檢測數據集的聚類。合理調用Spark RDD算子實現(xiàn)模糊聚類，把可緩存到內存的中間結果做Cache，降低中間讀寫磁盤操作的時間，且通過廣播聚類中心方式高效分發(fā)變量，減少節(jié)點間網絡通信消耗。另外，對比了KDD CUP 99數據集分別在單機和集群環(huán)境下的檢測率和運行時間，并對比數據集預處理前后的檢測率和聚類效率。應用結果表明，所設計的并行化方法提高了FCM算法的速度，對KDD CUP 99數據集的預處理降低了實驗數據的維度，提高了準確度和處理速度。