張發(fā)揚，李玲娟，陳 煜

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

VFDT算法基于Storm平臺的實現(xiàn)方案

張發(fā)揚，李玲娟，陳 煜

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

以提升流數(shù)據的分類效率為目標，研究如何在流數(shù)據處理平臺Storm上實現(xiàn)快速決策樹算法-VFDT。設計了VFDT基于Storm的分布式并行化實現(xiàn)方案，將VFDT算法分為建樹、分類和評價共三個模塊，建樹模塊負責決策樹的初始化和增量建樹，分類模塊負責對待分類樣本進行分類標記，評價模塊負責用已標記的樣本對VFDT決策樹進行評價。通過正確設計Storm拓撲中的Spout/Bolt實現(xiàn)各模塊的功能，通過為分類Bolt設定多個Task來實現(xiàn)分類模塊的并行化;用內存數(shù)據庫Redis實現(xiàn)三個模塊的有效銜接和決策樹的保存;用消息中間件Kafka來提高算法對流數(shù)據突增的容忍度。基于該方案的VFDT算法實現(xiàn)與測試結果表明:在Storm集群環(huán)境下的VFDT算法分類效率相對于單機環(huán)境有顯著提高，而且合理設定分類Bolt的Task數(shù)可使分類效率進一步提高。

流數(shù)據;快速決策樹算法;分布式;并行化;Storm

0 引言

20世紀末，為適應網絡監(jiān)控、入侵檢測、情報分析、商業(yè)交易管理和分析等應用的要求，數(shù)據流技術應運而生［1］。數(shù)據流中的數(shù)據(即流數(shù)據)是有序的、連續(xù)的且不斷變化的，甚至是無限的［2］，不能像傳統(tǒng)的靜態(tài)數(shù)據一樣將其存儲在硬盤或者內存之中，即再次處理這些數(shù)據的代價將非常昂貴。

流數(shù)據挖掘是指從快速、大量、連續(xù)的數(shù)據流中挖掘出有價值的信息。數(shù)據流具有快速、連續(xù)、大量的特點，傳統(tǒng)數(shù)據挖掘(Data Mining，DM)算法難以對流數(shù)據進行有效的挖掘。對于流數(shù)據，它的搜集和挖掘過程必須同時進行，且必須以最快的速度從不斷到來的數(shù)據中挖掘出用戶所需要的信息。傳統(tǒng)的靜態(tài)數(shù)據挖掘通常能滿足數(shù)據分析處理的精確性要求，但是，對流數(shù)據而言，由于數(shù)據收集的時間和處理速度有限，因此得到的挖掘模型是近似結果。流數(shù)據挖掘結果的近似性是其不同于傳統(tǒng)數(shù)據挖掘的一個重要特點。

分類挖掘是數(shù)據挖掘的主要任務之一，決策樹算法是分類挖掘的一類主流算法。VFDT(Very Fast Decision Tree)算法［3］是經典的流分類算法之一。VFDT在假設數(shù)據流不發(fā)生概念漂移的情況下對持續(xù)不斷的數(shù)據流進行增量學習，能夠很好地適應流數(shù)據的分類。VFDT算法可以使每條訓練樣本的處理花費恒定的內存和時間，因此可以有效解決時間和內存的限制問題。該算法通過最初的少量樣本生成隨時可用的分類器，并可隨著訓練樣本的到來，對原有分類器進行增量更新，不斷優(yōu)化原始決策樹，即VFDT算法以增量的形式，通過不斷地將葉子節(jié)點替換為決策節(jié)點而生成決策樹。

與傳統(tǒng)的靜態(tài)大數(shù)據處理平臺Hadoop［4］不同，Storm是開源的流數(shù)據處理框架［5］，能夠高效可靠地處理源源不斷的數(shù)據流。流挖掘算法運行于數(shù)據流處理平臺是充分發(fā)揮算法效力的前提。為此，文中研究如何將VFDT算法部署到Storm平臺上進行分布式并行化實現(xiàn)，以提高VFDT算法對流數(shù)據的分類效率。

1 VFDT算法分析

VFDT算法是通過對Hoeffding樹改進而實現(xiàn)的。Hoeffding樹是通過不斷地將葉子節(jié)點轉換為內部節(jié)點而生成的，其中每個葉節(jié)點都存有關于屬性值的統(tǒng)計信息，這些統(tǒng)計信息用于計算屬性的信息增益。當數(shù)據流中一個新的樣本到來后，該樣本沿著決策樹從上到下遍歷，樹的每個內部節(jié)點都對其進行劃分測試，根據不同的屬性取值，樣本進入不同的分枝，最終到達樹的葉節(jié)點。當數(shù)據到達葉節(jié)點后，更新該葉節(jié)點上的統(tǒng)計信息。如果統(tǒng)計信息的計算結果滿足節(jié)點分裂條件，則該葉節(jié)點變?yōu)閮炔抗?jié)點，并產生基于該內部節(jié)點的子女葉節(jié)點。子女葉節(jié)點的個數(shù)取決于新的內部節(jié)點的屬性的可能取值數(shù)目。

VFDT算法采用信息熵或者Gini指標作為選擇分裂屬性的標準，以Hoeffding不等式作為判定節(jié)點分裂的條件。選擇Hoeffding不等式作為節(jié)點分裂條件的目的是確定葉子節(jié)點變?yōu)閮炔抗?jié)點所需要的樣本數(shù)目，以達到使用盡量少的樣本建立準確率較高的決策樹的目的。

以t作時間戳，xt表示t時刻到達的數(shù)據向量，數(shù)據流可表示為{…，xt-1，xt，xt+1，…}［6］。VFDT算法的有關定義如下:

(1)信息增益［7］:葉子節(jié)點l中存儲訓練樣本集D的統(tǒng)計信息，則對于樣本集D分類所需的期望信息為Info(D)=-pilog2(pi)。其中，pi是樣本集D中任意一條樣本屬于類Ci的概率，pi= Ci，D/D，m是類別屬性的取值個數(shù)。對于葉子節(jié)點可能的分裂屬性A，設A有v個取值，則利用屬性A對樣本集D進行劃分的期望信息為InfoA(D)=-，屬性A的信息增益為Gain(A)=Info(D)-InfoA(D)。

(2)Hoeffding邊界［8-9］:對一個真值隨機變量r∈R，設對 r取了 n個獨立的觀察值，平均值為 r-，其Hoeffding約束以1-δ的概率保證變量r的真實值與觀察值r-之差小于ε，即:P(r≥r--ε)=1-δ。其中，ε，r代表信息增益，R的取值范圍是log2#Classes(Classes是類別屬性取值的數(shù)量)。

(3)主動分裂系數(shù)τ［10］:τ的作用在于當幾個屬性的信息增益值G幾乎相等時，可能需要更多的樣本來決定出葉子節(jié)點的決策屬性，通過設定τ值來主動選擇屬性并實現(xiàn)葉子節(jié)點分裂。當滿足ΔG＜ε＜τ時，選擇ΔG中信息增益最大或者次大的屬性作為該葉子節(jié)點的決策屬性。

VFDT算法的建樹流程如圖1所示。

概括地說，一條訓練樣本是一個Tuple，即一條流，樣本中各屬性的元素與初始化階段抽取的屬性信息保持一致，通過分析流的非類別屬性與類別屬性的關系建立一棵決策樹。增量建樹過程就是不斷將葉子節(jié)點轉化為內部節(jié)點的過程，其中每個葉子節(jié)點都將保存有關節(jié)點分裂的統(tǒng)計信息。當一個訓練樣本到達之后，從根節(jié)點開始，根據該節(jié)點的屬性取值進入不同的分支，以此過程進行遞歸直至到達葉子節(jié)點。到達葉節(jié)點之后將對葉子節(jié)點的統(tǒng)計信息進行更新，當統(tǒng)計值滿足計算的閾值時將觸發(fā)計算各可能屬性的信息增益以及Hoeffding邊界值，若滿足節(jié)點分裂的條件，則將該葉子節(jié)點轉化為內部節(jié)點，并根據決策屬性的取值產生新的葉子節(jié)點。

2 Storm平臺

Apache Storm是由Twitter公司開源的分布式實時計算系統(tǒng)。與Hadoop的批處理相比，Storm具有更好的實時性、可拓展性和容錯性，能有效地處理流數(shù)據，被廣泛用于實時分析、在線機器學習等場景［11］。

圖1 VFDT算法的建樹流程

在Storm內部，數(shù)據流是由拓撲(Topology)來處理的。拓撲包含Spout、數(shù)據源以及Bolt［12］。Bolt是拓撲的一個重要實體，負責數(shù)據流動轉換，比如計算、過濾、聚合以及機器學習等。一個拓撲可以有一個或者多個Bolt實現(xiàn)數(shù)據流的復雜流轉。

Storm集群的基本架構如圖2所示，主要包括兩種節(jié)點:主節(jié)點Nimbus(Master Node)以及工作節(jié)點Supervisor(Worker Node)。其中，Nimbus既負責將代碼分發(fā)到不同的工作節(jié)點，又負責監(jiān)控集群。在每個工作節(jié)點上都會運行一個Supervisor，負責監(jiān)聽Nimbus分配給該節(jié)點的工作［13］。每個Worker進程執(zhí)行一個具體的Topology，Worker中的執(zhí)行線程為Executor，每個Executor中又包含一個或者多個Task，Task為Storm的最小處理單元。一個運行中的Topology是由運行在一臺或者多臺工作節(jié)點上的Worker來完成具體的業(yè)務操作。Nimbus與Supervisor之間的通信由Zookeeper來完成。

圖2 Storm集群的基本架構

3 VFDT算法基于Storm的實現(xiàn)方案設計

文中設計的VFDT基于Storm的分布式并行化實現(xiàn)方案，將VFDT算法分為建樹、分類和評價共三個模塊，建樹模塊負責決策樹的初始化和增量建樹，分類模塊負責對待分類樣本進行分類標記，評價模塊負責用已標記樣本對VFDT決策樹進行評價。各模塊都有相應的Topology，如圖3所示。三個模塊之間通過內存數(shù)據庫Redis［14］實現(xiàn)銜接，從而形成一個整體的計算框架，Redis也用于決策樹的保存;消息中間件Kafka［15］用來提高算法對流數(shù)據突增情況的容忍度。

圖中的TraData表示外部數(shù)據源，為建樹模塊提供訓練樣本;Dspout1作為建樹拓撲的數(shù)據源從TraData中拉取數(shù)據并傳遞給其他數(shù)據處理Bolt;DataPro Bolt主要工作是對訓練樣本進行初始化，并將其轉換成算法所需要的類;VFDT Bolt接收樣本，并利用訓練樣本進行決策樹的增量建立;VFDTPrint Bolt接收增量建立的決策樹，并將決策樹進行序列化存儲到Redis數(shù)據庫中;ClaData表示外部數(shù)據源，為分類模塊提供待分類樣本;Kafka是消息中間件，訂閱ClaData中的樣本，同時供分類模塊進行消費;KafkaSpout作為分類拓撲的數(shù)據源，接收Kafka中的待分類樣本，并將樣本進行分發(fā);Tree Spout1表示拓撲的決策樹數(shù)據源，從Redis中實時獲取決策樹并進行分發(fā);Classification Bolt將利用Tree Spout1傳遞到決策樹對待分類樣本進行分類; InstPrint Bolt主要是對標記好的樣本進行存儲;EvaData表示外部數(shù)據源，為評價模塊提供評價樣本;Tree Spout2的功能與Tree Spout1一致;Evaluation Bolt利用EvaData對Tree Spout2中的決策樹進行評價;ResPrint Bolt將對評價結果進行存儲。

(1)建樹模塊。

如前所述，建樹模塊主要負責決策樹的初始化以及決策樹的增量建立。初始化的過程主要包括數(shù)據集屬性信息的抽取以及根節(jié)點的建立。決策樹的增量建立過程包括讀入訓練樣本集和基于圖1所示的流程建立決策樹。如圖3所示，DSpout1作為訓練樣本的數(shù)據源，不斷向負責數(shù)據預處理的DataPro Bolt發(fā)送訓練樣本，DataPro Bolt將訓練樣本處理成算法需要的類，并將其傳遞到負責建樹的VFDT Bolt中，VFDT Bolt將調用VFDT插入樣本的方法實現(xiàn)決策樹的動態(tài)更新，最后將VFDT決策樹傳遞到VFDTPrint Bolt中實現(xiàn)決策樹的序列化并存儲到Redis中。

(2)分類模塊。

分類模塊的主要功能是完成對待分類樣本的標記工作?？紤]到待分類樣本數(shù)量大且源源不斷地到來，當數(shù)據源突然增加時，有可能導致算法在Storm上并發(fā)度不足而引起異常，文中使用了消息中間件Kafka做數(shù)據暫存區(qū)。Kakfa具有高性能、高拓展性、分布式及持久性，當數(shù)據源突然增加時可以將部分數(shù)據持久化至硬盤中去，不至于造成數(shù)據的丟失［15］。為保證分類模塊的拓撲能夠保持較高的數(shù)據吞吐量，文中將該模塊中的數(shù)據預處理DataPro Bolt以及對待分類樣本進行分類的Classification Bolt的Task都設置為多個，以提高處理的并發(fā)度。

如圖3所示，DataPro Bolt使用Shuffle Grouping(隨機分組)的流分組方式從KafkaSpout中拉取數(shù)據，使得該Bolt的多個Tasks中的每個Task處理的樣本數(shù)目大致相同。Classification Bolt同樣采用Shuffle Grouping的方式使該Bolt中每個Task能夠平均處理數(shù)據。為了使該Bolt中的每個Task可以取到相同的決策樹，這部分還采用All Grouping(廣播分組)方式從負責由Redis內存數(shù)據庫中實時獲取 VFDT決策樹的 Tree Spout1中拉取決策樹。最后對Classification Bolt中標記過的待分類樣本采用Global Grouping(全局分組)的方式發(fā)送到InstancePrint Bolt中。

其中，Classification Bolt中利用決策樹VFDTTree對待分類樣本 ClaData進行分類的偽代碼如算法1所示。

(3)評價模塊。

評價模塊的主要功能是利用已標記的評價樣本實現(xiàn)對實時傳遞過來的VFDT決策樹的評價。為了保證評價樣本的實時性，文中采用滑動窗口的方式保存最新的評價數(shù)據，每隔一秒觸發(fā)一次評價，并將評價結果傳輸至ResultPrint Bolt中。

如圖3所示，Dspout2作為評價樣本的數(shù)據源，向DataPro Bolt發(fā)送數(shù)據，經過DataPro Bolt處理后發(fā)送到負責評價的Evaluation Bolt中，在Evaluation Bolt中維持一個固定大小的滑動窗口，用于存儲最近的N條評價數(shù)據。每當Evaluation Bolt從Tree Spout2中拉取最新的決策樹后，都利用窗口中的樣本對決策樹進行一次評價，并將評價結果發(fā)送到ResPrint Bolt中，實現(xiàn)結果的存儲。

4 VFDT算法基于Storm的實現(xiàn)與性能測試

文中分別在單機和集群環(huán)境下，用Java進行了VFDT算法的實現(xiàn)，算法運行環(huán)境是:

集群硬件環(huán)境:1個Nimbus節(jié)點，2個Supervisor節(jié)點。

集群軟件環(huán)境:操作系統(tǒng)為Centos6.4、JRE1.7.0_ 13、Zookeeper-3.4.6、Storm0.9.1、Kafka2.8.1、redis-2.4.5。

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

目的是借助流處理平臺提高VFDT算法的效率。為了驗證所設計的VFDT算法基于Storm的實現(xiàn)方案的可行性和有效性，分別測試了單機與集群環(huán)境下，基于Storm的VFDT算法分類模塊的吞吐量與分類Bolt (Classification Bolt)的Task線程數(shù)(即并行度)的關系，以及相同的Task線程數(shù)下數(shù)據處理時間與數(shù)據量的關系。

測試使用的數(shù)據集是KDD CUP的Nursery數(shù)據集，屬性個數(shù)是8，類別屬性取值個數(shù)是5，基本訓練樣本數(shù)量是12 400，通過復制得到所需量的分類樣本。

(1)吞吐量測試。

實驗通過復制Nursery得到大規(guī)模的分類樣本。單機與集群環(huán)境下，VFDT分類模塊對應于不同分類線程(Task)數(shù)的吞吐量如圖4所示。

可以看出，單機環(huán)境下，Task數(shù)為8時，吞吐量達到最大，為35 106.7條/s，當線程繼續(xù)增加，吞吐量呈現(xiàn)下降趨勢。集群環(huán)境下，Task數(shù)為8時，吞吐量達到最大，為88 007.5條/s，當線程繼續(xù)增加，吞吐量略呈下降趨勢。

(2)數(shù)據處理時間測試。

圖5對比了當單機和集群環(huán)境下 Classification Bolt的Task數(shù)為8時，不同數(shù)據量所需的處理時間。單機環(huán)境下，隨著數(shù)據量的增加，處理時長急劇增加;而集群環(huán)境下，隨著數(shù)據量的增加，處理時長緩慢增加。

(3)測試結果分析。

由圖4以及圖5可以看出，基于 Storm集群的VFDT算法在處理規(guī)模較大的流式數(shù)據時，吞吐量優(yōu)勢明顯，對數(shù)據量的增加具有較高的承受力。這是由于Storm是將Topology的各個組件(Spout/Bolt)分配到不同的Executor中，并在多個Worker中執(zhí)行的。由圖4還可以看出，合理設定分類Bolt的Task數(shù)可以最大限度發(fā)揮Storm的并行處理能力。

5 結束語

文中將經典的流數(shù)據分類挖掘算法-VFDT部署于流數(shù)據處理平臺Storm上，以實現(xiàn)對流數(shù)據的分布式并行化分類。所設計的VFDT算法基于Storm的實現(xiàn)方案按算法功能劃分出建樹模塊、分類模塊和評價模塊，其中的分類模塊以并行化方式運作;通過合理配置Storm拓撲和使用Redis與Kafka提高了實現(xiàn)方案的完整性和可行性。基于該方案的VFDT算法實現(xiàn)與性能測試結果說明了方案的正確性和有效性，也說明了基于Storm的VFDT算法對大規(guī)模流數(shù)據有良好的適應能力。

［1］ 史金成，胡學鋼.數(shù)據流挖掘研究［J］.計算機技術與發(fā)展，2007，17(11):11-14.

［2］ 顧 偉.分布式流數(shù)據實時計算框架的研究和開發(fā)［D］.杭州:浙江理工大學，2013.

［3］ Gama J，Rocha R，Medas P.Accurate decision trees for mining high-speed data streams［C］//Proceedings of the ninth ACM SIGKDD international conference on knowledge discovery and data mining.［s.l.］:ACM，2003:523-528.

［4］ White T.Hadoop:the definitive guide［M］.［s.l.］:O’Reilly Media，Inc.，2012.

［5］ The Apache Foundation.Storm official website［EB/OL］. ［2014-04-08］.http://storm-project.net.

［6］ Raahemi B，Zhong W，Liu J.Peer-to-peer traffic identification by mining IP layer data streams using concept-adapting very fast decision tree［C］//Proc of 20th IEEE international conference on tools with artificial intelligence.［s.l.］:IEEE，2008:525-532.

［7］ Maron O，Moore A W.Hoeffding races:accelerating model selection search for classification and function approximation ［J］.Advances in Neural Information Processing Systems，1993，6(1):59-66.

［8］ 王 濤，李舟軍，胡小華，等.一種高效的數(shù)據流挖掘增量模糊決策樹分類算法［J］.計算機學報，2007，30(8):1244-1250.

［9］ Littlestone N.Learning quickly when irrelevant attributes abound:a new linear-threshold algorithm［J］.Machine Learning，1988，2(4):285-318.

［10］蔣良孝，蔡之華，劉 釗.一種基于信息增益的分類規(guī)則挖掘算法［J］.中南大學學報:自然科學版，2003，34(z1):69-71.

［11］Github Inc.Storm Wiki［EB/OL］.［2013-12-07］.https:// github.com/nathanmarz/storm/wiki.

［12］Marz N.Storm:distributed and fault-tolerant real time computation［EB/OL］.［2011-10-21］.https://www.infoq.com/ presentations/Storm-Introduction.

［13］Petkov V，Gerndt M.Integrating parallel application development with performance analysis in periscope［C］//Proc of IPDPSW.［s.l.］:IEEE，2010:1-8.

［14］曾泉勻.基于Redis的分布式消息服務的設計與實現(xiàn)［D］.北京:北京郵電大學，2014.

［15］Kreps J，Narkhede N，Rao J.Kafka:a distributed messaging system for log processing［C］//Proceedings of the NetDB. Athens，Greece:［s.n.］，2011:1-7.

Implementation Scheme of VFDT Algorithm on Storm Platform

ZHANG Fa-yang，LI Ling-juan，CHEN Yu
(School of Computer，Nanjing University of Posts and Telecommunications，Nanjing 210003，China)

In order to improve the classification efficiency of the stream data，studies how to implement VFDT algorithm on Storm，a stream data processing platform.A scheme of distributed parallel implementing of VFDT algorithm based on Storm platform is designed. The VFDT algorithm is divided into three modules including building tree module，classification module and evaluation module.The building tree module is responsible for the initializing and incremental building of decision tree，and the classification module for classifying the samples，and the evaluation module for evaluating the VFDT decision tree using the labeled samples.The functions of each module are realized by correctly designing the Spout/Bolt of Storm Topology，and the parallelization of the classification module is implemented by deploying multiple tasks for Classification Bolt.The memory database Redis is used to realize the effective connection of the three modules and the preservation of the decision tree.The message middleware Kafka is used to improve the tolerance of burst stream data.The results of implementing and testing VFDT algorithm based on the proposed scheme show that the classification efficiency of VFDT algorithm under the Storm cluster environment is significantly improved compared with that under the single machine environment，and the classification efficiency can be further improved by reasonably setting the task number in Classification Bolt.

stream data;Very Fast Decision Tree(VFDT);distribution;parallelization;Storm

TP311

A

1673-629X(2016)09-0192-05

10.3969/j.issn.1673-629X.2016.09.043

2015-11-13

2016-03-03< class="emphasis_bold">網絡出版時間:

時間:2016-08-23

國家自然科學基金資助項目(61302158，61571238);中興通訊產學研項目

張發(fā)揚(1990-)，男，碩士研究生，CCF會員，研究方向為流數(shù)據挖掘;李玲娟，教授，CCF會員，研究方向為數(shù)據挖掘、信息安全、分布式計算。

http://www.cnki.net/kcms/detail/61.1450.TP.20160823.1359.044.html