基于Storm的分布式實(shí)時數(shù)據(jù)流密度聚類算法

牛麗媛，張桂蕓

（天津師范大學(xué)計(jì)算機(jī)與信息工程學(xué)院，天津300387）

隨著網(wǎng)絡(luò)的不斷發(fā)展以及大數(shù)據(jù)時代的到來，海量的數(shù)據(jù)分析變得尤為重要，數(shù)據(jù)流聚類是大數(shù)據(jù)處理的關(guān)鍵技術(shù).流數(shù)據(jù)具有連續(xù)、實(shí)時、高維、有序等不同于傳統(tǒng)數(shù)據(jù)集的特點(diǎn).這要求數(shù)據(jù)流聚類算法要具備實(shí)時處理增量數(shù)據(jù)的能力；能夠挖掘數(shù)據(jù)流中任意形狀的簇；能夠處理高速數(shù)據(jù)流，并要降低時間、空間復(fù)雜度，有效處理數(shù)據(jù)流中的噪聲.Aggarwal等[1]在2003年提出的CluStream經(jīng)典框架使用在線微聚類對數(shù)據(jù)流進(jìn)行初始聚類，并按金字塔時間框架存儲，按離線宏聚類對用戶查詢做出響應(yīng).文獻(xiàn)[2]提出的基于密度的空間數(shù)據(jù)流在線聚類算法OLDStream，在先前聚類結(jié)果上對增量空間數(shù)據(jù)進(jìn)行聚類，但該算法僅通過對新增空間點(diǎn)及其滿足核心點(diǎn)條件的鄰域數(shù)據(jù)做局部聚類，來實(shí)現(xiàn)空間數(shù)據(jù)流的在線聚類，不能體現(xiàn)數(shù)據(jù)流聚類的實(shí)時性.文獻(xiàn)[3]提出的DRCluStream算法對CluStream進(jìn)行了改進(jìn)，將流數(shù)據(jù)的在線微聚類部分拆分成局部和全局2個部分做分布式計(jì)算，但由于在線微聚類過程中使用K-means聚類方法，需要用戶指定聚類的簇數(shù)，且算法使用基于距離的度量準(zhǔn)則，聚類結(jié)果均趨于球形，無法挖掘任意形狀的簇.文獻(xiàn)[4]提出了DBSCAN增量式聚類算法，該算法對新增數(shù)據(jù)聚類后，按照規(guī)定邏輯將結(jié)果合并到已有數(shù)據(jù)中，從而避免了對已有數(shù)據(jù)進(jìn)行“二次聚類”，但它只對單個增量數(shù)據(jù)的合并邏輯進(jìn)行了處理，并沒有考慮到批量數(shù)據(jù)的合并，因此對于較大級別的增量數(shù)據(jù)，其增量合并的效率較低.基于以上研究的不足，本文提出一種分布式混合數(shù)據(jù)流聚類算法DBS-Stream.該算法在局部節(jié)點(diǎn)使用CluStream經(jīng)典框架，利用DBSCAN初始化數(shù)據(jù)，得到非球形聚類結(jié)果，從而克服了CluStream框架對非球形和噪聲數(shù)據(jù)聚類效果不佳的缺點(diǎn)，同時，對多個局部聚類結(jié)果進(jìn)行增量合并，在一定程度上彌補(bǔ)了DBSCAN算法的不足，在中心節(jié)點(diǎn)采用基于密度的聚類算法，對全局進(jìn)行再次聚類，同時在Storm分布式環(huán)境下實(shí)現(xiàn)算法，避免了提前確定K值，因此可有效優(yōu)化聚類效率和通信代價.

1 基本概念和算法

1.1 基本概念

定義1對空間中一點(diǎn)p及距離r，以p為中心、r為半徑的區(qū)域內(nèi)數(shù)據(jù)點(diǎn)的個數(shù)稱為點(diǎn)p關(guān)于距離r的密度，記為 D（p，r）.

定義2對空間中任一點(diǎn)p、距離r，及給定閾值pmin，若 D（p，r）≥pmin，則稱 p 為核心點(diǎn).

定義3對核心點(diǎn)q，以該點(diǎn)為中心、r為半徑的圓形區(qū)域{p∈D|dist（p，q）≤r}稱為核心點(diǎn)q的鄰域，記為Nr（q），其中dist（p，q）表示p、q之間的距離.

定義4對于核心點(diǎn)q，點(diǎn)p∈Nr（q），而點(diǎn)p為非核心點(diǎn)，稱p為q所在簇的邊界點(diǎn)，不在任何簇中的點(diǎn)稱為孤立點(diǎn).

定義5給定一列空間點(diǎn)p1，p2，…，pn，若pi直接密度可達(dá)于 pi+1（i=1，2，…，n-1），則稱 p1密度可達(dá)于pn.

定義6帶有時間戳T1，T2，…，TN的d維點(diǎn)集X1，…，XN，稱（2d+3）元組為微簇，其中：和分別表示數(shù)據(jù)的一階矩和二階矩，CF2t表示時間戳的平方和，CF1t表示時間戳的和，n表示微簇中的數(shù)據(jù)點(diǎn)個數(shù).

定理若核心點(diǎn)p分別屬于2個簇C1和C2，則C1和C2可密度相連為一個簇.

證明設(shè)當(dāng)前數(shù)據(jù)集為D.因?yàn)楹诵狞c(diǎn)p∈C1，所以任意屬于簇C1的空間點(diǎn)o都密度相連于點(diǎn)p，又由于任意密度相連于點(diǎn)p的空間點(diǎn)都密度可達(dá)于核心點(diǎn)p，所以任意空間點(diǎn)o∈C1都密度可達(dá)于點(diǎn)p.同理，任意空間點(diǎn)o′∈C2也都密度可達(dá)于點(diǎn)p.因此o∈C1和o′∈C2密度相連，所以C1和C2可密度相連為一個簇.

定理的結(jié)論說明若核心點(diǎn)p分別屬于多個聚類簇，則可以合并這些簇為一個簇，因此核心點(diǎn)最終僅屬于一個簇.

1.2 CluStream經(jīng)典兩段式框架

CluStream是經(jīng)典的兩段式流聚類框架.一段是在線微聚類過程，首先利用K-means算法初始化微簇，并將結(jié)果按金字塔時間幀結(jié)構(gòu)儲存，然后提取實(shí)時流入的數(shù)據(jù)流特征進(jìn)行增量維護(hù)微簇，通過為每個微簇定義一個閾值，來判斷新到達(dá)的數(shù)據(jù)點(diǎn)屬于某個已有的簇或需要創(chuàng)建一個新簇.若需要創(chuàng)建一個新簇，為了保持q值不變，需要刪除最近最少使用的簇或者對2個已有的簇進(jìn)行合并.同時，利用離線宏聚類對用戶查詢作出響應(yīng).另一段是離線部分，根據(jù)用戶給定的時間范圍h和期望的宏簇數(shù)目k，對不同時間粒度的聚類結(jié)果進(jìn)行查詢.將當(dāng)前時間點(diǎn)tc的快照減去tc-h的快照，得到最近時間范圍h內(nèi)的快照N（tc，h），然后將N（tc，h）視為加權(quán)虛擬點(diǎn)，利用Stream算法進(jìn)行聚類，得到時間范圍h內(nèi)的數(shù)據(jù)流聚類結(jié)果.

1.3 DBSCAN算法

DBSCAN是典型的密度聚類算法.其基本思想是尋找相鄰區(qū)域內(nèi)密度可達(dá)的部分，將其聚成一類.它的優(yōu)勢是可以利用類的高密度連通性快速發(fā)現(xiàn)任意形狀的類.首先從數(shù)據(jù)庫對象集D中任取一點(diǎn)p，并給定半徑r和數(shù)據(jù)點(diǎn)閾值pmin；然后確定D中從p關(guān)于半徑r密度可達(dá)的所有數(shù)據(jù)點(diǎn)的鄰域.如果p是核心點(diǎn)，則可找到一個關(guān)于r和pmin的類；如果p是一個邊界點(diǎn)，則p被暫時標(biāo)注為孤立點(diǎn).

2 分布式實(shí)時數(shù)據(jù)流密度聚類算法DBS-Stream

2.1 算法基本思想

DBS-Stream算法采用CluStream兩段式流數(shù)據(jù)處理框架和DBSCAN算法，并將CluStream的在線微聚類過程分為局部節(jié)點(diǎn)聚類（生成局部微簇）與中心節(jié)點(diǎn)聚類（利用局部微簇對全局微簇進(jìn)行增量更新）2部分，這種做法可有效解決內(nèi)存消耗過大的問題，進(jìn)而降低時間復(fù)雜度.

DBS-Stream算法具體步驟為

（1）中心節(jié)點(diǎn)初始化全局微簇.

（2）在一個單位時間內(nèi)，局部節(jié)點(diǎn)接收待挖掘的數(shù)據(jù).

（3）到達(dá)下一個單位時間后，利用局部節(jié)點(diǎn)處理數(shù)據(jù)，生成局部微簇結(jié)果，并發(fā)送結(jié)果到中心節(jié)點(diǎn).

（4）判斷局部微簇是否能合并到全局微簇，若能則合并，若不能則將局部微簇作為新微簇加入到全局微簇.

（5）刪除全局微簇中權(quán)重不滿足條件的微簇，并更新金字塔存儲快照.

在局部節(jié)點(diǎn)聚類中，為克服K-means算法的局限性和缺點(diǎn)，使用DBSCAN代替K-means進(jìn)行聚類，產(chǎn)生局部微簇.具體過程為，數(shù)據(jù)流進(jìn)入每個局部節(jié)點(diǎn)后，在單位時間t內(nèi)使用DBSCAN算法對局部節(jié)點(diǎn)累積的數(shù)據(jù)進(jìn)行聚類，得到局部節(jié)點(diǎn)的核心點(diǎn)、邊界點(diǎn)和孤立點(diǎn).為了節(jié)省內(nèi)存開銷，局部節(jié)點(diǎn)采用滑動窗口模型，對先到達(dá)的數(shù)據(jù)設(shè)定較低權(quán)重，而后到達(dá)的數(shù)據(jù)設(shè)定較高權(quán)重.

在中心節(jié)點(diǎn)聚類中，為盡量降低時間復(fù)雜度，將孤立點(diǎn)舍棄.中心節(jié)點(diǎn)利用局部節(jié)點(diǎn)發(fā)送的核心點(diǎn)與邊界點(diǎn)對全局微簇進(jìn)行增量更新.具體過程為，判斷局部節(jié)點(diǎn)微簇的核心點(diǎn)是否與全局微簇的核心點(diǎn)直接密度可達(dá)，若是，則將二者合并，同時更新簇內(nèi)數(shù)據(jù)的平均到達(dá)時間，若否，則生成新的微簇并添加到全局微簇結(jié)果；然后根據(jù)時間權(quán)重在全局微簇結(jié)果中移除過期的微簇；最后存儲更新后的全局微簇結(jié)果，等待下一次增量更新[5].

2.2 算法描述

算法實(shí)現(xiàn)的偽代碼如下.

初始化數(shù)據(jù)：中心節(jié)點(diǎn)根據(jù)初始化數(shù)據(jù)進(jìn)行DBSCAN聚類，獲得中心節(jié)點(diǎn)初始聚類結(jié)果：由核心點(diǎn)集合P、邊界點(diǎn)集合Q、孤立點(diǎn)集合O組成.

在線部分：

局部節(jié)點(diǎn)在線聚類：

Input：新到達(dá)的數(shù)據(jù)P，時間間隔t.

Output：局部節(jié)點(diǎn)DBSCAN之后的微簇結(jié)果（核心點(diǎn)集合，邊界點(diǎn)集合，孤立點(diǎn)集合，當(dāng)前時間）.

Begin：

Repeat：

t時間間隔內(nèi)，獲得到達(dá)的數(shù)據(jù)點(diǎn)集合P；對數(shù)據(jù)集合P進(jìn)行DBSCAN聚類；通過聚類結(jié)果得到微簇.發(fā)送聚類結(jié)果到全局節(jié)點(diǎn).

End

全局節(jié)點(diǎn)增量合并：

Input：局部節(jié)點(diǎn)發(fā)送來的局部聚類結(jié)果，微簇刪除周期T.

Output：全局增量合并后的聚類結(jié)果.

Begin：

得到局部節(jié)點(diǎn)發(fā)送的微簇；初始化合并結(jié)果記錄數(shù)組R.

Repeat：取得中心節(jié)點(diǎn)的微簇.

Repeat：獲取局部節(jié)點(diǎn)微簇Xi.

IF（Xi可以和 Yi合并）

{記錄Yi可以合并Xi，將這條結(jié)果保存到R}

Until遍歷所有局部節(jié)點(diǎn)的微簇.

Until遍歷所有中心節(jié)點(diǎn)的微簇.

根據(jù)合并結(jié)果記錄數(shù)組R，合并局部微簇到中心節(jié)點(diǎn)微簇，同時更新每個微簇的時間.

沒有被合并到中心節(jié)點(diǎn)的微簇各自作為一個新的微簇結(jié)果，合并進(jìn)中心節(jié)點(diǎn)微簇.遍歷中心節(jié)點(diǎn)微簇.

IF（當(dāng)前時間 tc微簇的 t＜ T）

{刪除該微簇}

根據(jù)微簇結(jié)果更新金字塔結(jié)構(gòu).

End

離線部分：DBSCAN滿足用戶查詢.

3 在Storm平臺上的實(shí)現(xiàn)方案

3.1 Storm部署

Storm[6]是一個實(shí)時的、分布式的具備高容錯的計(jì)算系統(tǒng).Storm的核心組件[7]中Nimbus是主節(jié)點(diǎn)，負(fù)責(zé)資源分配和任務(wù)調(diào)度；Supervisor是配置的從節(jié)點(diǎn)，負(fù)責(zé)接受Nimbus分配的任務(wù)，并啟動worker用以運(yùn)行具體處理組件邏輯的進(jìn)程.worker中每一個spout/bolt的線程稱為一個task.

在Storm中，需要設(shè)計(jì)一個用于實(shí)時計(jì)算的圖狀結(jié)構(gòu)，即拓?fù)?這個拓?fù)鋾惶峤唤o集群，由集群中的主控節(jié)點(diǎn)（master node）分發(fā)代碼，將任務(wù)分配給工作節(jié)點(diǎn)（workernode）執(zhí)行.一個拓?fù)渲邪╯pout和bolt兩種角色，spout發(fā)送消息，bolt則負(fù)責(zé)轉(zhuǎn)換這些數(shù)據(jù)流，并完成計(jì)算、過濾等操作.

3.2 DBS-Stream算法在Storm平臺上的拓?fù)鋱D

DBS-Stream算法在Storm平臺上的拓?fù)鋱D見圖1.圖1中，SpoutA接收初始化數(shù)據(jù)，將其發(fā)送到DBSCAN Blot；SpoutB通過KAFKA[8]接收實(shí)時到來的待處理數(shù)據(jù)，并發(fā)送到Local Bolt；SpoutC做時間計(jì)算；SpoutD接收初始化參數(shù)，將其發(fā)送到DBSCAN Blot；Spout E通過KAFKA接收用戶的查詢參數(shù)，并發(fā)送到Macro Clustering Bolt；DBSCAN Bolt以SpoutA和SpoutD傳送來的初始化數(shù)據(jù)和參數(shù)進(jìn)行初始化全局微簇.Local Bolt屬于分布式部分，到達(dá)的數(shù)據(jù)會平均分配到每一個Local Bolt線程.當(dāng)接收到SpoutC傳送來的時間信息后，在Local Bolt上生成局部微簇，并將結(jié)果發(fā)送到Redis進(jìn)行保存.Global Bolt實(shí)現(xiàn)全局微簇的增量更新，SpoutC傳送時間信息后，Global Bolt開始取出Redis中的數(shù)據(jù)進(jìn)行全局微簇結(jié)果的增量合并.Macro Clustering Bolt接收SpoutE傳送來的用戶查詢參數(shù)，對Global Bolt中的全局結(jié)果進(jìn)行聚類查詢，并將查詢結(jié)果發(fā)送到Print Bolt.

圖1 DBS-Stream在Storm上的拓?fù)鋱DFig.1 DBS-Stream′s topology on Storm

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

4.1 實(shí)驗(yàn)環(huán)境與數(shù)據(jù)集

算法實(shí)驗(yàn)在實(shí)驗(yàn)室小集群上實(shí)現(xiàn)，集群中設(shè)有1個Nimbus節(jié)點(diǎn)和3個Supervisor節(jié)點(diǎn)，軟件環(huán)境由Jdk-1.8.0-32bit、Zookeeper-3.4.6、Storm-0.9.1、KAFKA-2.9.1、Redis-2.4.5等構(gòu)成，操作系統(tǒng)為centos6.4.實(shí)驗(yàn)采用的數(shù)據(jù)集是經(jīng)過預(yù)處理的酵母菌數(shù)據(jù)集[9]，描述預(yù)測蛋白質(zhì)的細(xì)胞定位點(diǎn)包含1 486條實(shí)例數(shù)據(jù)，數(shù)據(jù)分布狀況為：CYT（細(xì)胞質(zhì)或細(xì)胞骨架）463條；NUC（核）429條；MIT（線粒體）244條；ME3（膜蛋白，無N末端信號）163條；ME2（膜蛋白，未切割的信號）51條；ME1（膜蛋白，裂解信號）44條；EXC（細(xì)胞外）37條；VAC（液泡）30條；POX（過氧化物酶體）20條；ERL（內(nèi)質(zhì)網(wǎng)腔）5條.實(shí)驗(yàn)通過每秒隨機(jī)抽取該數(shù)據(jù)集中的100條數(shù)據(jù)來模擬數(shù)據(jù)流.數(shù)據(jù)集共含9維屬性，其中1個是序列名稱，另外8個預(yù)測屬性分別是mcg、gvh、alm、mit、erl、pox、vac和 nuc.

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

對DBS-Stream和CluStream在Storm上處理相同數(shù)據(jù)集的聚類精度（clutering accuracy）進(jìn)行對比，結(jié)果見圖2.由圖2可見，本算法的聚類質(zhì)量高于CluStream，而且在數(shù)據(jù)量不是很大的時候明顯高于CluStream.DBS-Stream通過在局部節(jié)點(diǎn)進(jìn)行局部微簇聚類，可以過濾大部分噪聲數(shù)據(jù)，同時在全局節(jié)點(diǎn)合并時會根據(jù)時間權(quán)重有效處理權(quán)重低的微簇，也進(jìn)一步保證了聚類質(zhì)量.

對于相同的數(shù)據(jù)流，DBS-Stream和CluStream產(chǎn)生的通信代價（communication cost）見圖3.由圖3可見，DBS-Stream的通信代價要明顯低于CluStream.因?yàn)榕cCluStream相比，DBS-Stream會對聚類結(jié)果中無效的噪聲點(diǎn)進(jìn)行過濾，避免了無效數(shù)據(jù)的傳輸，從而較CluStream節(jié)省較多的通信資源，因而在通信代價上具有明顯的優(yōu)勢.噪聲點(diǎn)的存在也會降低聚類的質(zhì)量和效率，因此DBS-Stream對噪聲點(diǎn)的處理既降低了通信代價，又提高了聚類的質(zhì)量和效率.

圖2 DBS-Stream算法和CluStream的聚類精度Fig.2 Clustering accuracies of DBS-Stream and CluStream

圖3 DBS-Stream算法和CluStream的通信代價Fig.3 Communication costs of DBS-Stream and CluStream

圖4給出了依據(jù)測試結(jié)果得出的線程處理壓力（threadhandlingpressure）與線程個數(shù)（numberofthread）的關(guān)系.由圖4可見，Local Bolt線程的處理壓力隨著并行線程數(shù)的增加而降低，Global Bolt線程的處理壓力隨著并行線程的增加呈凈增加的趨勢.該測試結(jié)果與理論一致.

圖4 線程處理壓力與線程個數(shù)的關(guān)系Fig.4 Relationship between thread handling pressure and number of thread

利用數(shù)據(jù)集單位時間產(chǎn)生固定的數(shù)據(jù)條數(shù)來模擬數(shù)據(jù)流，分別用CluStream和DBS-Stream處理相同的數(shù)據(jù)流，處理所用時間t（processing time）見圖5.由圖5可見，DBS-Stream與CluStream的執(zhí)行時間均隨數(shù)據(jù)流的增長呈線性增長趨勢.DBS-Stream處理時間略長，這是因?yàn)镈BSCAN算法的復(fù)雜度比K-means高，但DBSCAN的優(yōu)勢在于可對任意形狀的簇進(jìn)行聚類.當(dāng)數(shù)據(jù)條數(shù)較大時，二者處理時間相差不大，不超過10%，屬于可接受范圍.總體上說，相對于DBS-Stream在聚類精度和通信代價方面的優(yōu)化，在處理時間上稍做犧牲是值得的.

圖5 DBS-Stream和CluStream的處理時間Fig.5 Processing time of DBS-Stream and CluStream

5 結(jié)束語

基于Storm平臺的DBS-Stream算法與CluStream算法相比，在聚類質(zhì)量及通信代價方面均有較好的表現(xiàn)，并且DBS-Stream算法可以處理任意形狀的數(shù)據(jù)流，在聚類結(jié)果的形狀上沒有偏倚，而且不需根據(jù)經(jīng)驗(yàn)確定K值，也就無需對數(shù)據(jù)形成先驗(yàn)知識.

[1]AGGARWAL C C，HAN J W，WANG J Y，et al.A framework for clustering evolving data streams[C]//Proceedings of the 29th VLDB Conference，Berlin，2003.

[2]于彥偉，王沁，鄺俊，等.一種基于密度的空間數(shù)據(jù)流在線聚類算法[J].自動化學(xué)報，2012，38（6）：1051-1059.YU Y W，WANG Q，KUANG J，et al.A density based spatial data flow online clustering algorithm[J].Acta Automatica Sinica，2012，38（6）：1051-1059（in Chinese）.

[3]馬可，李玲娟.分布式實(shí)時流數(shù)據(jù)聚類算法及其基于Storm的實(shí)現(xiàn)[J].南京郵電大學(xué)學(xué)報（自然科學(xué)版），2016，36（2）：104-110.MA K，LI L J.Distributed real-time streaming data clustering algorithm and its implementation based on Storm[J].Journal of Nanjing University of Posts and Telecommunications（Natural Science Edition），2016，36（2）：104-110（in Chinese）.

[4]田路強(qiáng).基于DBSCAN的分布式聚類及增量聚類的研究與應(yīng)用[D].北京：北京工業(yè)大學(xué)，2016.TIAN L Q.DBSCAN-based Distributed Clustering and Incremental Clustering Research and Application[D].Beijing：Beijing University of Technology，2016（in Chinese）.

[5]高宏賓，侯杰，劉勁飛.分布式密度和中心點(diǎn)數(shù)據(jù)流聚類算法的研究[J].計(jì)算機(jī)應(yīng)用與軟件，2013，30（10）：181-184.GAO H B，HOU J，LIU J F.Research on distributed density and centralized data flow clustering algorithm[J].Computer Application and Software，2013，30（10）：181-184（in Chinese）.

[6]陳東明，劉健，王冬琦，等.基于MapReduce的分布式網(wǎng)絡(luò)數(shù)據(jù)聚類算法[J].計(jì)算機(jī)工程，2013，39（7）：76-82.CHENG D M，LIU J，WANG D Q，et al.A distributed network data clustering algorithm based on MapReduce[J].Computer Engineering，2013，39（7）：76-82（in Chinese）.

[7]Apache.Storm：Distributed and fault-tolerant realtime computation[EB/OL].http：//storm.Apache.org，2015-09-18/2017-08-09.

[8]Apache.Apache Kafka：A high-throughput，distributed，publish-subscribe messaging system[EB/OL].http：//kafka.Apache.org，2015-09-18/2017-08-12.

[9]UCI.Kenta Nakai.Datasets yeast[EB/OL].http：//archive.ics.uci.edu/ml/datasets/Yeast，2015-09-19/2017-09-12.