莊 榮，李玲娟

(南京郵電大學(xué) 計算機學(xué)院,江蘇 南京 210023)

0 引 言

面對實時到達、連續(xù)、無限的流數(shù)據(jù)[1]，傳統(tǒng)的數(shù)據(jù)挖掘算法難以滿足流數(shù)據(jù)挖掘的需求，因而流數(shù)據(jù)的分類挖掘[2]研究一直是熱門話題并且具有重大意義。概念適應(yīng)快速決策樹算法(concept-adapting very fast decision tree，CVFDT)[3]是經(jīng)典的流分類算法之一。CVFDT主要克服了VFDT(very fast decision tree)算法[4]對于數(shù)據(jù)樣本的不斷變化而不能更換模型的缺點，并且可以有效地解決概念漂移[5]的問題。與傳統(tǒng)的靜態(tài)大數(shù)據(jù)處理平臺Hadoop[6]不同，Spark[7]擴展了廣泛使用的MapReduce[8]模型，提出了基于內(nèi)存的并行計算框架，通過將中間結(jié)果緩存在內(nèi)存中以減少I/O磁盤操作，從而更高效地支持多次迭代式計算模式。為此，文中研究了基于Spark的CVFDT分類算法并行化，用以提高CVFDT算法對流數(shù)據(jù)的分類效率。

1 CVFDT算法分析

1.1 CVFDT簡介

CVFDT屬于一種增量式的分類挖掘方法，即用新到樣本修正舊分類器，產(chǎn)生新分類器，以適應(yīng)新環(huán)境。CVFDT在樹的所有節(jié)點上維持統(tǒng)計信息用于計算基于屬性值的信息增益測試，即統(tǒng)計測試，并基于Hoeffding不等式確定葉節(jié)點變成分支節(jié)點所需的樣本數(shù)目，對數(shù)據(jù)流建立分類決策樹。

以t為時間戳，xt表示t時刻到達的數(shù)據(jù)向量，數(shù)據(jù)流可表示為{…,xt-1，xt，xt+1,…}[9]。CVFDT算法的有關(guān)定義如下：

1.2 CVFDT算法流程

在拿到新的數(shù)據(jù)流樣本后，從上到下遍歷決策樹，并在樹的每個分支節(jié)點根據(jù)屬性取值等判斷進入不同的分支，最終到達樹的葉節(jié)點。隨著數(shù)據(jù)流樣本的不斷增多，信息增益測試為了滿足一定條件，其葉節(jié)點必須要以較高的置信度確定最佳劃分屬性，從而變成一個分支節(jié)點，這樣循環(huán)可以不斷地決策學(xué)習(xí)新的葉節(jié)點。如遇到概念漂移問題，CVFDT就會在相應(yīng)分支節(jié)點上并行生成備選子樹，原子樹會隨著備選子樹的精度遠遠超過其本身時被替換和釋放。

2 Spark平臺

Spark不同于Hadoop MapReduce的是，Job中間輸出結(jié)果可以保存在內(nèi)存中，從而不再需要頻繁讀寫HDFS[12]，可以顯著提高運行速度。Spark還提供了SparkSQL、Spark Streaming[13]、MLib[14]等計算模式組件，更適用于分布式平臺場景。Spark Streaming是構(gòu)建在Spark上的處理Stream數(shù)據(jù)的框架，基本原理是將Stream數(shù)據(jù)分成小的時間片斷(幾秒)，以類似batch批量處理的方式來處理這小部分數(shù)據(jù)。

彈性分布式數(shù)據(jù)集RDD[15]是Spark的核心和基礎(chǔ)。RDD是一種分布式的內(nèi)存抽象，表示只讀的分區(qū)記錄的集合。它只能通過在穩(wěn)定物理存儲中的數(shù)據(jù)集或其他已有的RDD上執(zhí)行一些確定性操作(并行操作中的轉(zhuǎn)換操作)來創(chuàng)建，并且RDD僅支持粗粒度轉(zhuǎn)換，即在大量的記錄上執(zhí)行單一的操作，因此省去了大量的磁盤I/O操作，對于需要多次迭代計算的機器學(xué)習(xí)算法、交互式數(shù)據(jù)挖掘來說，效率得到了極大地提升；同時它具有非常出色的容錯機制和調(diào)度機制，能夠有效確保系統(tǒng)的長時間穩(wěn)定運行。所以Spark能夠更高效地支持交互式查詢以及迭代式計算等多種計算模式。

3 CVFDT基于Spark的并行化方案設(shè)計

3.1 CVFDT的分割點計算過程的并行化

在CVFDT建樹過程中計算最佳分割點時，需要將Hoeffding邊界作為節(jié)點分裂條件找到最佳分割點，其首要任務(wù)是計算并比較各個屬性的最佳基尼分割指數(shù)。在面對含多種屬性類別的數(shù)據(jù)集時，線性串行的計算模式會大大降低運行效率。

因此，針對CVFDT算法的分割點計算過程，考慮到每個屬性的基尼分割指數(shù)求解是完全獨立計算，可以對這些計算進行同步，設(shè)計了如圖1所示的屬性間并行化方案。

圖1 針對分割點計算過程的屬性間并行化方案

如圖1所示，首先計算每個任務(wù)的最佳基尼指數(shù)和Hoedding邊界，從而找到每個任務(wù)的最佳分割點；然后在每個任務(wù)計算完成后進行比較，獲取最佳分割點。這種改進的計算模式可以有效地降低時間復(fù)雜度。

3.2 基于Spark的RDD實施CVFDT的并行化

1.Spark的并行計算過程簡述。

RDD為了讓海量數(shù)據(jù)分散在不同的計算節(jié)點上進行并行處理，會橫向地拆分成N個分區(qū)，因此對RDD進行計算操作時，集群會對每個分區(qū)進行計算，然后由相應(yīng)的集群控制器將結(jié)果進行匯總，最終統(tǒng)計整個RDD結(jié)果。因此，Spark的并行屬于“橫向”并行化。

2.針對建樹過程的基于Spark的橫向并行化。

在3.1提出的屬性間并行化的基礎(chǔ)上，基于Spark的橫向并行化，針對CVFDT算法的建樹過程做如下并行化改造：

(1)If(都是同類屬性的數(shù)值||獲取的屬性列表個數(shù)不超過閾值)

(2)將含有同屬性最多數(shù)值的類復(fù)制給節(jié)點N的Decision類并且返回；

(3)Else

(4)得到節(jié)點N的屬性列表AttList，將所有屬性列表轉(zhuǎn)化為對應(yīng)的RDD；

(5)計算由每個RDD生成的并行化任務(wù)，匯總并比較每個最佳分割點；

(6)再計算Hoeffding邊界產(chǎn)生節(jié)點分裂條件，找到最佳分割點；

(7)在AttList中找到該分割點相應(yīng)屬性的屬性列表并刪除，然后對其余屬性表進行分裂，得到屬性表Attlist1,Attlist2,…,AttlistN；

(8)新的子節(jié)點N1,N2,…,Nn由節(jié)點N生成,并將屬性列表Attlist1,Attlist2,…,AttlistN分別賦給N1,N2,…,Nn；

(9)執(zhí)行buildTree(n1)，buildTree(n2)，…，buildTree(n)操作，遞歸建立決策樹。

4 實驗結(jié)果與分析

選擇傳統(tǒng)單機CVFDT算法和基于Spark集群的CFVDT并行化算法對實驗數(shù)據(jù)集進行分類操作，算法運行環(huán)境是：

集群硬件環(huán)境：1個Master節(jié)點，2個Slave節(jié)點。

集群操作系統(tǒng)：centos6.6。

集群軟件環(huán)境：JRE1.7.0_13、Scala_2.11.6、Spark1.3.1。

單機環(huán)境：eclipse_4.5.0、JRE1.7.0_13、Windows7、2.13 GHz、4 GB內(nèi)存。

目的是借助流數(shù)據(jù)處理平臺提高CVFDT算法的執(zhí)行效率。為了驗證所設(shè)計的CVFDT算法基于Spark的并行化方案的可行性和有效性，對比了單機和集群并行環(huán)境下，CVFDT算法處理不同數(shù)據(jù)量所需的時間。

實驗使用的數(shù)據(jù)集源自Kaggle比賽，基于美國UGC網(wǎng)站Stumbleupon提供的歷史數(shù)據(jù)，設(shè)計分類模型，預(yù)測該網(wǎng)站提供的網(wǎng)頁是否長期流行。訓(xùn)練集樣本數(shù)目為10 706個，測試集樣本大小為5 171 M。

4.1 建樹效率的比較

為了考慮算法的實用性，選取了200k、300k、500k、800k、1 000k條這五種不同規(guī)模的數(shù)據(jù)集，測試結(jié)果如表1所示。

表1 建樹時間測試結(jié)果

數(shù)據(jù)大小/條 現(xiàn)有算法/s并行化算法/s綜合效率提高/%200k6766.90.13300k110106.43.37500k197187.15.28800k310289.67.011 000k401367.49.14

當(dāng)選取200k條規(guī)模數(shù)據(jù)集時，算法(建樹)效率提升微乎其微，并且10 000條數(shù)據(jù)規(guī)模下，建樹時間反而增加。原因是資源管理、網(wǎng)絡(luò)傳輸會伴隨著Spark運行集群而產(chǎn)生額外開銷。當(dāng)數(shù)據(jù)規(guī)模達到300k以上時，或者海量數(shù)據(jù)規(guī)模時，基于Spark的CVFDT并行化有明顯的效率提升。

4.2 數(shù)據(jù)處理時間的比較

圖2對比了單機和Spark集群環(huán)境下在數(shù)據(jù)規(guī)模分別為200 M、300 M、500 M、800 M、1 000 M時所需的時間。

圖2 對應(yīng)不同數(shù)據(jù)量的處理時間測試結(jié)果

在單機環(huán)境下，隨著數(shù)據(jù)規(guī)模的擴大，數(shù)據(jù)處理時間急劇增加；在Spark集群環(huán)境下，在200 M數(shù)據(jù)規(guī)模下，處理時間提升不明顯，除了上一節(jié)提到的原因之外，主要還存在求解每個分裂條件的基尼系數(shù)時，會依照分裂條件對RDD進行過濾，然后再調(diào)用count()函數(shù)來統(tǒng)計個數(shù)，每一次的count操作都是在Spark集群中完成，然后將計算結(jié)果傳輸給虛擬機。當(dāng)屬性列表數(shù)據(jù)條數(shù)N不是很大時，Spark的優(yōu)勢無法體現(xiàn)。300 M數(shù)據(jù)量之后，可以看到并行化算法的數(shù)據(jù)處理時間明顯減少，當(dāng)數(shù)據(jù)量到1 000 M時，處理時間縮減了66.6%。

4.3 測試結(jié)果分析

由表1和圖2可以看出，基于Spark集群的并行化CVFDT算法在處理規(guī)模較大的流式數(shù)據(jù)時，運行效率有所提高，并且在數(shù)據(jù)規(guī)模增大時，其效果會越發(fā)明顯。并且并行化CVFDT算法相對于單機環(huán)境在處理海量數(shù)據(jù)時處理效率有顯著提高，而且合理設(shè)定RDD過濾可使處理效率進一步提高。

5 結(jié)束語

將經(jīng)典的流數(shù)據(jù)分類挖掘算法CVFDT部署于流數(shù)據(jù)處理平臺Spark上，借助構(gòu)建在Spark之上的實時計算框架Spark Streaming來實現(xiàn)對流數(shù)據(jù)的并行化分類。對CVFDT算法進行了屬性間并行化改造，并且基于Spark的RDD進行了CFVDT算法在建樹流程上的橫向化并行。測試結(jié)果證明了該設(shè)計思想的正確性和方案的有效性，也說明了基于Spark的并行化CVFDT算法對大規(guī)模流數(shù)據(jù)有良好的適應(yīng)能力。

參考文獻：

[1] DING Shifei,WU Fulin,QIAN Jun,et al.Research on data stream clustering algorithms[J].Artificial Intelligence Review,2015,43(4):593-600.

[2] ABURROUS M,HOSSAIN M A,DAHAL K,et al.Predicting phishing websites using classification mining techniques with experimental case studies[C]//7th international conference on information technology.Las Vegas,NV,USA:IEEE,2010:176-181.

[3] 王 濤,李舟軍,顏躍進.一種基于哈希鏈表的高效概念漂移連續(xù)屬性處理算法[J].計算機工程與科學(xué),2008,30(8):65-68.

[4] 袁 磊,張 陽,李 梅,等.在數(shù)據(jù)流管理系統(tǒng)中實現(xiàn)快速決策樹算法[J].計算機科學(xué)與探索,2010,4(8):673-682.

[5] MINKU L L,WHITE A P,YAO Xin.The impact of diversity on online ensemble learning in the presence of concept drift[J].IEEE Transactions on Knowledge & Data Engineering,2010,22(5):730-742.

[6] DITTRICH J,QUIANéRUIZ J A,JINDAL A,et al.Hadoop++:making a yellow elephant run like a cheetah (without it even noticing)[J].Proceedings of the VLDB Endowment,2010,3(1-2):515-529.

[7] 黎文陽.大數(shù)據(jù)處理模型Apache Spark研究[J].現(xiàn)代計算機,2015(8):55-60.

[8] 沈 超,鄧彩鳳.論Storm分布式實時計算工具[J].中國科技縱橫,2014(3):53.

[9] RAAHEMI B,ZHONG Weicai,LIU Jing.Peer-to-peer traffic identification by mining IP layer data streams using concept-adapting very fast decision tree[C]//Proceedings of the 20th IEEE international conference on tools with artificial intelligence.Dayton,OH,USA:IEEE,2008.

[10] 張發(fā)揚,李玲娟,陳 煜.VFDT算法基于Storm平臺的實現(xiàn)方案[J].計算機技術(shù)與發(fā)展,2016,26(9):192-196.

[11] YIN Chunyong,FENG Lu,MA Luyu,et al.A feature selection algorithm of dynamic data-stream based on Hoeffding inequality[C]//International conference on advanced information technology and sensor application.[s.l.]:IEEE,2015:92-95.

[12] 歐陽永.運營商大數(shù)據(jù)系統(tǒng)建設(shè)的分析與研究[D].南京:南京郵電大學(xué),2016.

[13] 管祥青.大數(shù)據(jù)可視化模型的協(xié)同過濾算法研究及應(yīng)用[D].長沙:湖南大學(xué),2015.

[14] XIN R S,GONZALEZ J E,FRANKLIN M J,et al.GraphX:a resilient distributed graph system on Spark[C]//International workshop on graph data management experiences and systems.New York,NY,USA:ACM,2013.

[15] 劉志強,顧 榮,袁春風(fēng),等.基于SparkR的分類算法并行化研究[J].計算機科學(xué)與探索,2015,9(11):1281-1294.