基于Spark的CVFDT分類算法并行化研究

莊 榮，李玲娟

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

0 引 言

面對實(shí)時(shí)到達(dá)、連續(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ù)樣本的不斷變化而不能更換模型的缺點(diǎn)，并且可以有效地解決概念漂移[5]的問題。與傳統(tǒng)的靜態(tài)大數(shù)據(jù)處理平臺Hadoop[6]不同，Spark[7]擴(kuò)展了廣泛使用的MapReduce[8]模型，提出了基于內(nèi)存的并行計(jì)算框架，通過將中間結(jié)果緩存在內(nèi)存中以減少I/O磁盤操作，從而更高效地支持多次迭代式計(jì)算模式。為此，文中研究了基于Spark的CVFDT分類算法并行化，用以提高CVFDT算法對流數(shù)據(jù)的分類效率。

1 CVFDT算法分析

1.1 CVFDT簡介

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

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

1.2 CVFDT算法流程

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

2 Spark平臺

Spark不同于Hadoop MapReduce的是，Job中間輸出結(jié)果可以保存在內(nèi)存中，從而不再需要頻繁讀寫HDFS[12]，可以顯著提高運(yùn)行速度。Spark還提供了SparkSQL、Spark Streaming[13]、MLib[14]等計(jì)算模式組件，更適用于分布式平臺場景。Spark Streaming是構(gòu)建在Spark上的處理Stream數(shù)據(jù)的框架，基本原理是將Stream數(shù)據(jù)分成小的時(shí)間片斷(幾秒)，以類似batch批量處理的方式來處理這小部分?jǐn)?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操作，對于需要多次迭代計(jì)算的機(jī)器學(xué)習(xí)算法、交互式數(shù)據(jù)挖掘來說，效率得到了極大地提升；同時(shí)它具有非常出色的容錯機(jī)制和調(diào)度機(jī)制，能夠有效確保系統(tǒng)的長時(shí)間穩(wěn)定運(yùn)行。所以Spark能夠更高效地支持交互式查詢以及迭代式計(jì)算等多種計(jì)算模式。

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

3.1 CVFDT的分割點(diǎn)計(jì)算過程的并行化

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

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

圖1 針對分割點(diǎn)計(jì)算過程的屬性間并行化方案

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

3.2 基于Spark的RDD實(shí)施CVFDT的并行化

1.Spark的并行計(jì)算過程簡述。

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

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

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

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

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

(3)Else

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

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

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

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

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

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

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

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

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

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

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

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

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

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

4.1 建樹效率的比較

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

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

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

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

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

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

4.3 測試結(jié)果分析

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

5 結(jié)束語

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

參考文獻(xiàn)：

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

[4] 袁 磊,張 陽,李 梅,等.在數(shù)據(jù)流管理系統(tǒng)中實(shí)現(xiàn)快速決策樹算法[J].計(jì)算機(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)代計(jì)算機(jī),2015(8):55-60.

[8] 沈 超,鄧彩鳳.論Storm分布式實(shí)時(shí)計(jì)算工具[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ā)揚(yáng),李玲娟,陳 煜.VFDT算法基于Storm平臺的實(shí)現(xiàn)方案[J].計(jì)算機(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] 歐陽永.運(yùn)營商大數(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] 劉志強(qiáng),顧 榮,袁春風(fēng),等.基于SparkR的分類算法并行化研究[J].計(jì)算機(jī)科學(xué)與探索,2015,9(11):1281-1294.