VFDT算法基于Storm平臺的實現方案

張發揚，李玲娟，陳 煜

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

VFDT算法基于Storm平臺的實現方案

張發揚，李玲娟，陳 煜

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

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

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

0 引言

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

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

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

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

1 VFDT算法分析

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

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

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

(1)信息增益［7］:葉子節點l中存儲訓練樣本集D的統計信息，則對于樣本集D分類所需的期望信息為Info(D)=-pilog2(pi)。其中，pi是樣本集D中任意一條樣本屬于類Ci的概率，pi= Ci，D/D，m是類別屬性的取值個數。對于葉子節點可能的分裂屬性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是類別屬性取值的數量)。

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

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

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

2 Storm平臺

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

圖1 VFDT算法的建樹流程

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

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

圖2 Storm集群的基本架構

3 VFDT算法基于Storm的實現方案設計

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

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

(1)建樹模塊。

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

(2)分類模塊。

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

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

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

(3)評價模塊。

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

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

4 VFDT算法基于Storm的實現與性能測試

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

集群硬件環境:1個Nimbus節點，2個Supervisor節點。

集群軟件環境:操作系統為Centos6.4、JRE1.7.0_ 13、Zookeeper-3.4.6、Storm0.9.1、Kafka2.8.1、redis-2.4.5。

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

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

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

(1)吞吐量測試。

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

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

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

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

(3)測試結果分析。

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

5 結束語

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

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

［2］ 顧 偉.分布式流數據實時計算框架的研究和開發［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］ 王 濤，李舟軍，胡小華，等.一種高效的數據流挖掘增量模糊決策樹分類算法［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］蔣良孝，蔡之華，劉 釗.一種基于信息增益的分類規則挖掘算法［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的分布式消息服務的設計與實現［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);中興通訊產學研項目

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

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