面向流數據的決策樹分類算法并行化

季一木 張永潘 郎賢波 張殿超 王汝傳,2

1(南京郵電大學計算機學院 南京 210023)2(江蘇省無線傳感網高技術研究重點實驗室(南京郵電大學) 南京 210023)3(南京郵電大學先進技術研究院 南京 210023)4 (高維信息智能感知與系統教育部重點實驗室(南京理工大學) 南京 210094)

面向流數據的決策樹分類算法并行化

季一木1,2,3,4張永潘1郎賢波1張殿超1王汝傳1,2

1(南京郵電大學計算機學院 南京 210023)2(江蘇省無線傳感網高技術研究重點實驗室(南京郵電大學) 南京 210023)3(南京郵電大學先進技術研究院 南京 210023)4(高維信息智能感知與系統教育部重點實驗室(南京理工大學) 南京 210094)

(jiym@njupt.edu.cn)

隨著云計算、物聯網等技術的興起，流數據作為一種新型的大數據形態廣泛存在于電信、互聯網、金融等領域.與傳統靜態數據相比，大數據環境下的流數據具有快速、連續和隨時間變化等特點.同時數據流的隱含分布變化會帶來概念漂移問題.為了適應大數據環境下流數據分類算法的要求，必須對傳統的靜態離線數據分類算法進行改進，提出基于分布式計算平臺Storm的P-HT并行化算法.算法在滿足Storm流處理平臺要求基礎上，通過滑動窗口機制、替代子樹機制和并行化處理，提高了算法的靈活性和通用性，并且能良好地適應數據流的概念漂移.最后通過實驗驗證該算法的有效性和高效性，結果表明在與傳統C4.5算法相比精度沒有降低的情況下，改進的P-HT算法具有更大的吞吐量和更快的處理速度.

流數據；分類算法；Storm平臺；滑動窗口；C4.5算法;并行化算法

數據挖掘近年來正逐漸成為經濟學、人工智能等領域的研究熱點，在當前的大數據環境下，流式數據具有快速性、連續性、變化性和多樣性等特點[1]，數據流挖掘是數據挖掘領域的一個重要分支.各種流數據挖掘的技術和算法相繼被提出并得到運用.如何從連續不斷的數據流中挖掘出潛在的關聯和有用的信息，成為目前我們面臨的一項重要挑戰.與傳統的數據模型不同，數據流模型具有3個特性:1)數據到達速度快，實時性強;2)數據量巨大，難以將所有的數據都有效地存儲下來;3)數據一旦經過處理，除非需要保存供后續使用，否則不能被重復掃描，再次掃描數據消耗過大.然而傳統的數據挖掘方法在處理之前就將數據全部存儲記錄下來，進行分析時訪問存儲介質進行挖掘.但流數據環境下數據是快速到達的，且數據規模巨大，數據流的隱含分布變化會帶來概念漂移問題，傳統數據挖掘技術難以滿足數據流挖掘的要求[2].

分類是一種非常重要的數據挖掘技術，其目的是根據已有的數據集學習構造一個分類函數或分類模型，該分類模型能夠將新到樣本映射到一個具體的類別上.傳統的分類模型包括決策樹(decision tree)、貝葉斯算法(Bayes algorithm)、神經網絡(neural network)、K近鄰分類算法(K-nearest neighbour)、支持向量機(support vector machine)等[3].其中，決策樹模型[4]是最普遍的一種分類模型，相比與其他的分類模型，決策樹模型能簡單地生成可以理解的規則，同時計算量相對較小，而且可以處理多種數據類型.它是一種依托于策略抉擇而建立起來的樹，著眼于從一組無次序、無規則的實例中推理出以決策樹表示的分類規則.因而決策樹模型可以清晰地顯示出模型中的核心字段和潛在信息.

流數據的決策樹分類方法比傳統的分類在實時性和存儲限制等方面面臨更多的挑戰[5]，傳統數據分類方法，往往很難滿足數據流分類的需求和要求，因此需要將傳統分類模型進行重新調整.在文獻[6]中，Domingos等人提出的VFDT算法是一種流數據環境下的分類算法，VFDT基于Hoeffding不等式建立決策樹，每當新樣本流入時，VFDT都將該新樣本沿著決策樹從上到下遍歷，最終到達樹的葉子節點.Gama等人[7-8]在VFDT的基礎上提出VFDTc算法，使其能夠處理連續型數據.但是VFDTc算法對每一個可能的屬性值均計算信息熵，這在高速數據流環境下會大大加劇計算的負擔.Jin[9]同樣針對VFDT算法處理連續型屬性的局限，改進提出了NIP算法，NIP通過將連續屬性值劃分成不同的小片段，計算每個離散的小片段的信息熵，然后將有可能成為最優分割點的離散片段保留下來，將其余的片段刪除.將連續屬性離散化的方法，能夠顯著地縮小尋找最優分割點的范圍和計算量，但是如果錯誤地刪除了包含分割點的片段，整個決策樹的構建和分類效果都將受到很大的負面影響.Anagnostopoulos等人[10]提出一種基于概率估計法的流分類算法，但由于數據本身的影響，估計的結果會出現較大的偏差.

在數據流的隱含分布不斷變化的情況下，模型的準確率和穩定性不能得到保證，即發生了概念漂移問題.文獻[11]中通過時間序列分析方法對現有的處理概念漂移的策略進行了分類，并描述了自適應的學習過程.文獻[12]中，Gama提出時間滑動窗口機制，提高了模型訓練的實時性.文獻[13]提出的CVFDT算法是對VFDT算法的擴展，它保持了VFDT的速度和準確率，并且當數據流的隱含信息不斷變化時，它能更好地保證模型的實時性和準確度.Kuncheva等人[14]對流數據分類模型深入研究，發現模型的精確性不僅跟數據本身的質量和概念漂移發生的程度相關，滑動窗口的大小也會對準確率產生很大的影響.因此Kuncheva等人提出了基于可變滑動窗口的流數據分類方法.當數據流發生概念漂移時，滑動窗口能夠自適應地調整窗口大小，使得分類模型能夠更加有效地抵抗概念漂移，保證了分類模型的實時性和精確性.Liang等人[15]在CVFDT算法的研究基礎上，改進和擴展了CVFDT，提出puuCVFDT算法，該算法可以對沒有標簽和屬性值不確定的數據進行處理.與其類似，Zheng[16]同樣對CVFDT算法進行改進，提出了CFDT算法，該算法引入了一種新的樹型索引結構，首先采用聚類方法對數據進行預處理，進而對聚類簇進行掃描并抽取出隱含的信息，從而對樹的結構進行調整，這種方法使得CFDT樹既能提高模型的實時性，也保證了較高的分類準確率.表1對比了4種常見的分類算法.

Table 1 The Comparison of Common Classification Algorithms

表1中提到的經典C4.5算法[17]是由ID3發展而來的決策樹算法，相比于ID3算法，C4.5的特點主要在于C4.5用信息增益率來選擇屬性,且C4.5能夠完成對連續性數據的離散化處理，從而克服了ID3只能處理離散型數據的缺點.但是C4.5是針對離線的靜態數據集進行信息的挖掘.VFDT是基于Hoeffding樹的流分類算法，適應了大數據環境下的流數據分類要求，但是它不能處理連續性數據且當數據流發生概念漂移時，模型的精度會大大下降.VFDTc算法對VFDT所能處理的數據類型進行擴展，但是VFDTc算法對每一個可能的屬性值均計算信息墑，這大大加劇了計算負擔.CVFDT算法解決了流數據的概念漂移問題，且可以加入連續屬性的處理機制，但是對于決策樹的建立是串行化的方法，在高速的流數據環境下，模型訓練的速度不能達到要求.文獻[18]中提出了一種分布式的實時情感大數據流分析算法；文獻[19]提出了一種并行霍夫丁決策樹的方法，提高了實時決策樹分類算法的效率，但是并沒有解決流數據環境下的分類處理需要考慮有限內存、時間序列和單次處理的特點.Storm是Twitter開源的一個免費的分布式流計算系統，它能夠對持續大流量的數據流進行實時分析和快速響應，并且支持大數據流的并行化處理.

針對VFDT算法在流數據環境下不能解決概念漂移的問題，以及CVFDT算法對于大數據環境下的有限內存和訓練速度問題，結合分布式并行流處理平臺Storm在實時流數據處理上的快速性、實時性和安全性的優點，本文在CVFDT算法研究的基礎上，提出基于Storm平臺的實時P-HT并行化分類算法.算法引入可變滑動窗口，當發生概念漂移時，窗口及時地收縮，有利于較快地適應概念漂移，防止精度的大幅降低.同時提出了一種并行Hoeffding樹的方法，縮短了節點分裂時的計算時間，從而提高了決策樹模型訓練的速度.與CVFDT算法相比，P-HT有相同的精度和分類效率，但是加入Storm集群的并行化計算使得算法的建樹效率得到很大的提高.

1 CVFDT分類算法和Storm流處理平臺

1.1 CVFDT算法

CVFDT算法是一種基于Hoeffding不等式建立決策樹的方法，基于小樣本足以選擇最優的分裂屬性，使用Hoeffding邊界量化葉節點中確定最優分裂屬性所需要的樣本個數.

定義2. 信息增益.是用來衡量給定的屬性區分訓練樣例的能力，計算公式如式(1)(2)：

(1)

Gain(S,A)=Entropy(S)-

(2)

式(1)是信息熵的計算公式，其中pi是在樣例集S中不同類別Ci的樣本的比例.式(2)是信息增益的計算公式，其中Values(A)是屬性A所有可能值的集合，Sv是S中屬性A的值為v的子集(即Sv={s∈S|A(s)=v}).

CVFDT算法過程包括3個部分：CVFDTGrow過程、ForgetExample過程、CheckSplitValidity過程，其總體流程圖如圖1所示.CVFDT算法中的CVFDTGrow過程與Hoeffding Tree算法的生成樹類似，但是CVFDT通過保存每個節點上的統計數來檢驗原先的HT決策樹的有效性(而不是像VFDT只統計葉子節點).因為新的樣本不斷參與模型的生成導致HT樹不斷地生成和改變，刪除舊的樣本將出現困難.因此，在建立一個節點時會分配一個單獨的單調增加的ID.滑動窗口W是一個保存實時樣本的先進先出隊列，當新的樣本進入滑動窗口，隨之舊的樣本將要從窗口中滑出時，舊樣本所經過的所有內部節點的統計值Nijk減1，同時新的樣本將從根節點開始遍歷至最深的葉子節點，經過的所有節點統計值Nijk加1.

Fig. 1 The process of CVFDT algorithm圖1 CVFDT算法流程圖

CheckSplitValidity過程為：葉節點開始從數據流收集樣本，隨著樣本數量的增多，能夠以較高的置信度確定最佳劃分屬性時，則將該葉節點變成一個測試節點，然后對新的葉節點不斷地重復該學習過程.CVFDT維持一個訓練樣本的窗口，并通過在樣本進入和流出窗口時更新已學習的決策樹，使其與訓練樣本窗口保持一致.當一個新樣本到達之后，它將被加入到其所經過的所有決策樹節點；而當將一個樣本從決策樹中去除時，它也需要從所有受其影響的節點中移除，并且所有的統計測試都需要重新進行.當CVFDT懷疑有概念漂移發生時，它就并行在該節點生成一棵備選子樹.當備選子樹的精度遠大于原先的子樹時，原始的子樹被替換并釋放.

1.2 Storm實時處理平臺

Fig. 2 Storm on YARN cluster architecture圖2 Storm on YARN集群架構

Storm是Twitter支持開發的分布式、開源的、實時的流處理平臺.Storm集群采用基于YARN的管理模式，集群的架構為主從式(masterslaves)，由一個主節點和多個工作節點組成，Storm基于YARN的集群的架構如圖2所示：

其中，Nimbus和Supervisor都是快速失敗、無狀態的，所以某一個節點宕機立即重啟不會影響系統的運行，主節點和工作節點之間的協調是通過Zookeeper而完成的.

1) Nimbus.Nimbus是主節點，客戶端上傳的jar包會上傳到Nimbus，Nimbus進行代碼分發、任務分配、集群狀態監控等.

2) Zookeeper.負責集群的協調、共有數據的存放(如心跳信息)，主節點把任務分配信息寫到Zookeeper，各個工作節點會不時地從Zookeeper獲取自己的任務，某一個節點連接超時，則認為該節點失敗.

3) Supervisor.對應一臺物理節點，用于啟動worker.

Fig. 3 Storm stream grouping圖3 Storm數據流

在Storm中，一個實時的計算應用程序被封裝成一個任務拓撲(topology)，類似于Hadoop的MapReduce Job，Storm拓撲由Spout、Bolt組件和Streams組成.其中，Spout是整個topology的數據源，將數據發送給相應的Bolt；Bolt負責對接收到的數據進行計算，實現過濾、查詢等功能，可以級聯，也可以向外發送數據流.每個Spout，Bolt在集群中都是多線程運行的，消息的傳遞根據StreamGrouping完成，如圖3所示.一個實時應用的計算任務被打包成任務拓撲后發布，任務拓撲一旦提交后將會一直運行著，除非顯式去中止.數據流是Storm對數據進行的抽象，它是時間上的無窮的Tuple元組序列，數據流是通過流分組(stream grouping)所提供的不同策略實現在任務拓撲中流動.此外，為了滿足確保消息能夠且能被計算一次的需求，Storm還提供了事務任務topology.

2 基于Storm平臺的P-HT并行化算法

2.1 可變滑動窗口

滑動窗口模型隨著數據流中樣本的到達，新的數據不斷插入到滑動窗口中，窗口的時間差或者數據元素數量保持不變，因此可以保證基于滑動窗口中的樣本所生成的模型是隨著數據流的變化而不斷更新的，這保證了模型的實時性和準確性.通常窗口的大小在模型訓練中是固定不變的，這是由用戶根據經驗初始化的一個閾值.然而在數據流的隱含分布不斷變化的情況下，初始化的窗口大小值windowsize并不是最優的.初始化的值過大會導致模型不能較好地抵抗概念漂移，初始值太小會使得模型訓練樣本不充足而導致準確率下降.

因此本文結合Storm的分布式特點，提出一種實時的并行化窗口方案，基于Storm的并行化窗口方案初始化多個窗口將實時樣本流分為S1和S2，分別監測2條流的隱含信息分布，根據泊松過程和Hoeffding不等式得到式(3)：

Pr{X}≥(1+ε)E[X]≤

exp{-((1+ε)ln(1+ε)-ε)E[X]}，

(3)

其中，E(X)是樣本流的期值，ε是需要檢測概念漂移的極限.根據泰勒級數式(4)和泊松過程期望的式(5)：

(4)

(5)

Fig. 4 The process of sliding window圖4 滑動窗口示意圖

2.2 垂直并行化

決策樹的構建是一個反復迭代的過程，用傳統的串行方法來建樹，規模很小的數據量也需要花費大量的時間和資源，大數據環境下的流分類算法更是需要有效地提高模型的訓練速度，減小資源消耗.為了解決流數據分類算法面臨的有限內存和訓練效率問題，最有效的方式是采用分治法，將原有的計算任務分解成若干個相同的子任務來處理，使得每個計算機節點均衡負載.根據分治法思想，結合決策樹的生長過程，可以提出3種并行化的策略：任務并行化、水平并行化和垂直并行化.

Fig. 5 Vertical parallelization of P-HT圖5 P-HT垂直并行化原理

1) 任務并行化的思想是各個樹節點之間的并行.當內部節點進行分裂后將產生若干個不同的葉子節點，各個葉子節點再次進行分裂任務時，是不存在任何的依賴關系.這種并行方式不僅僅局限于兄弟節點之間，其他兄弟節點的孩子節點也是相同的并行性關系.但是這種方式的并行邏輯關系較為復雜，且使得各個計算機節點之間的通信帶變得很復雜，也加大了交互過程的資源消耗.

2) 水平并行化的思想是基于數據集的并行方法.將原始數據集劃分成N個子集，各個訓練樣本子集獨立地在不同的計算機節點上工作，對于每一個計算機節點上的數據都調用相同的生長過程，最終合并得到完整的模型.然而水平并行需要大量的可用內存，因為算法的復制模型需要在本地統計觀察，并且花費大量的時間來計算每個屬性的信息增益.

3) 在垂直并行中，計算機節點不存儲本地模型，每一個邏輯節點只存儲分發到的數據的統計信息，并且計算信息增益，最后將聚集本地的統計信息得到最終的決策樹.垂直并行更適合于具有很多屬性的實例,因為它把大部分的時間開銷在每個屬性的信息增益的計算上.相比于水平并行，它占用更少的內存，因為它不需要在每一個邏輯節點上復制模型.

因此本文提出基于垂直并行化思想的P-HT算法，并行化原理如圖5所示，每一個方框代表一個處理頁(processing item, PI)，方框里的數字代表并行度.Model-aggregator PI包含決策樹模型，它通過屬性流和控制流連接local-statistic PI，它通過屬性流發送葉子ID、屬性值、類別值等屬性內容事件至local-statistic PI，當節點需要分裂時，Model-aggregator PI通過控制流發送計算內容事件到local-statistic PI，每一個local-statistic PI通過計算所屬屬性的G(Xi)來確定最大和次大屬性xa，xb，然后將結果返回給Model-aggregatorPI，當所有的本地信息到達Model-aggregatorPI時，會通過計算Hoeffding不等式來決定是否進行分裂并將決策結果發送給evaluatorPI或者其目的PI.

2.3P-HT樹生長過程

定義3. 統計量Nijk.P-HT樹的每個內部節點都對經過的樣本信息進行統計，其中i代表屬性，j代表屬性取值，k代表類別.Nijk表示該節點對于每個屬性i的每個屬性值j，屬于k類的樣本的個數.

定義4. 訓練數據流.設D為數據流訓練樣本的集合，所有的樣本數據集被劃分成m個類{C1，C2,…,Cm}，m為正整數.又設每個樣本用一個d維屬性向量Xi=(x1i,x2i,…,xdi)和一個類標號yi表示，其中x1i,x2i,…,xdi是第i個樣本在d個屬性A1,A2,…,Ad上的值.令N=|D|表示D中樣本數據集的成員個數.

每一個訓練樣本(x，y)進入窗口后，都會記錄下它所到達的最大葉子節點的ID，當窗口內樣本數達到閾值需要刪除舊的樣本時，通過減去樣本在P-HT樹中每個ID小于存儲的ID的節點計數來實現舍棄舊樣本及更新模型的目的.

P-HT算法初始化時會設置一個檢測有效性間隔，當樣本數達到檢測間隔時，對于每一個內部節點，P-HT統計接下來到達的m個測試樣本，用它們來比較在此節點下所有替代子樹的精度.如果最佳替代子樹對于測試樣本擁有更高的分類準確率，則替代子樹會取代原先的節點；如果替代子樹的精確度在一段時間內沒有提高，P-HT也會刪除沒有進展的替代樹.P-HT樹的完整生長過程偽代碼見算法1：

算法1. P-HT樹生長過程.

輸入：當前決策樹P-HT、樣本流E的樣本(x,y)、信息增益函數G(x,y)、預設的置信度δ、檢測增長需要的樣本數nmin、用戶預設ties值τ、屬性并行計算的并行度Pnum；

輸出：實時決策樹.

① 初始時P-HT是根節點root；

② for 樣本流E中的每一個訓練樣本(x,y) do

③ 樣本(x,y)沿著P-HT從上到下遍歷，樹的每個內部節點都對其進行劃分測試，根據樣本的每個屬性取值進入不同的分枝，最終到達一個葉子節點；

④ 樣本(x,y)經過的每一個節點對于樣本統計值Nijk加1；

⑤ 葉子節點中的大多數樣本為同一類則認為其類為l；

⑥ ifnlmodnmin=0且l中的樣本類不能由大多數樣本類決定then

⑦ 通過統計值Nijk，基于并行度Pnum將屬性合理地分配給多個線程計算屬性Xi的信息增益函數G(Xi)；

⑧ 記xa為G值最大的屬性，xb為G值次大的屬性；

⑨ 通過δ和nl計算Hoeffding 值ε=

或者ε<τthen

2.4 基于Storm的P-HT算法實現

為了使快速決策樹算法能夠適應Storm實時平臺的流數據處理，本文提出并實現了在Storm平臺上實現P-HT并行化算法，該算法在保證精確度不低于傳統靜態C4.5分類算法的情況下，有效地提升了算法的執行效率，并且可以根據用戶的需求靈活選擇滑動窗口的大小和并行度，大大提高算法的適用性和處理速度.

Storm計算平臺提供了Spout和Bolt編程接口，Spout組件作為流數據的輸入，Bolt組件作為流數據的處理邏輯.我們利用Kafka消息中間件產生實時數據流，在數據準備階段我們將訓練數據集和的屬性集先交給一個AttAllocateSpout進行讀取，并將訓練樣本集通過Kafka模擬成數據流的形式，在算法Topology中利用KafkaSpout接口接收從Kafka消息隊列中傳來的數據流，傳遞給TransformBolt.在TransformBolt中，數據流將從字符串類型被轉換成instacnce類型，這是weka提供的一種數據結構，樣本作為instance在Bolt之間高速傳遞并建立決策樹.P-HTBolt初始化樹根節點root和替代子樹集altNodes、初始化統計量Nijk、初始化窗口W.ParallelBolt根據AttAllocateSpout分配的并行度，將所有屬性均衡分配給Storm集群的每個物理節點，根據統計量Nijk計算每個屬性的信息增益G(Xi)并傳遞給CheckSplitBolt進行分裂條件判斷，Check-SplitBolt接收每個屬性的G(Xi)，找出最大和次大G(Xi)，并根據Hoeffding邊界條件執行分裂.基于Storm的P-HT算法實現架構圖如圖6所示：

Fig. 6 The architecture of P-HT algorithm based on Storm圖6 基于Storm的P-HT算法實現架構

在進行分類時，同樣利用Kafka消息中間件產生實時數據流，利用KafkaSpout接口接收Kafka消息隊列中傳來的待分類樣本，傳遞給ClassifyBolt進行分類，ClassifyBolt從Redis數據庫中讀取實時的分類器對待分類樣本進行分類，并將分類結果傳遞給EvaluateBolt進行準確率的計算.OutputBolt將分類結果和評價結果輸出到文件或者數據庫中供用戶分析.用戶可以通過Redis數據庫配置算法變量需求，可以通過在Redis數據庫中建立key值為windowsize的整數來控制算法中滑動窗口的大小；可以通過建立key值為Pnum的整數來控制算法的并行度；也可以設置替代子樹的檢測間隔和檢測持續樣本數等變量.算法在P-HTBolt中初始化時從Redis中讀取用戶設置的窗口大小windowsize和并行度Pnum以及檢測間隔.并行度Pnum和Storm集群的計算機節點數決定了屬性增益的計算所消耗的時間.基于Storm平臺的P-HT并行化算法偽代碼見算法2：

算法2. 基于Storm的P-HT算法.

輸入：KafkaSpout接收的訓練數據流S={〈X1,C1〉,〈X2,C2〉,…,〈Xi,Ci〉}、并行計算的并行度Pnum、窗口大小windowsize、檢測概念漂移的間隔checkinterval、檢測樹增長的樣本數nmin；

輸出：P-HT決策樹.

① TransformBolt接收KafkaSpout傳來的數據流，將數據流轉換成Instance類型的實例子；

② AttAllocateSpout解析訓練集屬性向量Attribute,并根據并行度Pnum平均分配屬性；

③ P-HTBolt實時接收實例，初始化樹根節點root和替代子樹集altNodes，初始化統計量Nijk,初始化窗口W；

④ ifnlmodnmin=0 then

⑤ 將當前決策樹classifier，當前達到分裂條件的節點傳給ParallelBolt；

⑥ ParallelBolt根據AttAllocateSpout分配的并行度，將所有屬性均衡分配給Storm集群的每個物理節點，根據統計量Nijk計算每個屬性的信息增益G(Xi)；

⑦ CheckSplitBolt接收每個屬性的G(Xi)，找出最大和次大G(Xi)，并根據Hoeffd-ing邊界條件執行分裂；

⑧ 對于數據流S中的每一條樣本((x,y)，ID)，通過P-HT樹和所有節點的替代子樹進行分類排序；

⑨ID為樣本所經歷的葉子集合中最大的葉子ID號,依次添加樣本((x,y)，ID)到窗口中；

⑩ end if

2.5 復雜度分析

基于Storm平臺的P-HT并行化分類算法充分利用了Storm集群的實時流處理系統優點，同時也利用決策樹算法生長的垂直并行化思想，大大提高了算法對流數據的處理效率和算法吞吐量，算法維持的時間滑動窗口W提高了模型的實時性，也提升了算法的準確率，同時加入垂直并行計算屬性增益的思想，對算法的模型訓練時間有了大大的縮減.在基于Storm的P-HT算法中，算法2所需的內存復雜度由決策樹中節點的個數M、屬性個數D、每個屬性值的最大值V和類的個數C決定.傳統的串行建樹VFDT算法的復雜度為O(mdvc),相比與傳統的串行建樹的算法VFDT，若VFDT計算N個屬性的信息增益所需時間為T,則當屬性個數D大于并行度Pnum時，在Storm集群上進行計算所需的時間為TPnum，即算法2所需的內存大小為O(mdvc)Pnum.

3 實驗結果與分析

為驗證所提出的基于Storm的P-HT并行化算法在保證算法精度的基礎上，大大提高了模型的訓練效率.本文采用2種數據來測試：1)通過Storm的Spout讀取文件產生數據流，來測試算法在不同UCI數據集下的準確率；2)通過Kafka模擬真實數據流，分別從算法的并行化效果、抗概念漂移性能分析、算法的吞吐量3個方面來分析算法在Storm平臺上的并行化效果.

3.1 算法的準確率

1) 實驗數據

本文在UCI機器學習數據集倉庫中隨機選擇7個categorical類型的分類數據集供P-HT算法在Storm集群上進行測試，提取數據的屬性集和數據集并分別存入集群上相應的文件夾供Spout讀取.并用數據挖掘軟件weka-3-7對C4.5分類器和Naive Bayes分類器進行測試，所使用的UCI數據集的基本情況如表2所示：

Table 2 Description of UCI Data

2) 實驗環境

① 軟件環境

Storm版本：apache-storm-0.9.1；

Redis版本：Redis-2.8.13 64位；

Java版本：JDK1.7.0_55.

② 硬件環境

Storm集群由master，node1～node4等5個物理節點組成，分別負責Storm中拓撲程序的控制和計算工作，其中控制節點為master，上面運行Storm的Nimbus進程，計算節點為node1-4，上面運行Storm的Supervisor后臺進程及算法運算的Worker進程.

3) 實驗結果

利用weka-3-7 Explorer內置的J48(即C4.5)分類器和Naive Bayes分類器對UCI數據集進行測試，同時將UCI數據集的屬性集和樣本集存入txt文件，放入Storm集群的主節點文件夾中，使用storm jar命令將封裝好的VFDT算法和P-HT算法jar包提交到Storm集群上運行拓撲，這里設置P-HT算法的并行度為4.通過PrintBolt輸出的日志文件得到算法的準確率，準確率的計算公式為

Accuracy=NcorrectNtotal，

(6)

其中,Ncorrect代表被分類器正確分類的樣本數，Ntotal代表參與分類的所有樣本數目.計算結果得出不同數據集的算法準確率實驗結果如表3所示：

Table 3 Accuracy of the Classifiers

通過實驗結果的對比可以看出，相比傳統的靜態數據集分類算法C4.5和Naive Bayes，P-HT算法的精度并沒有大幅的下降，基本都在可以接受的誤差范圍之內，證明傳統的決策樹分類算法在Storm上的實現是有效的；同時在高速的流數據環境下，相比于VFDT算法，P-HT算法的準確率要高，所以替代子樹機制和垂直并行化處理機制的應用對基于Storm的并行化P-HT算法的準確率有明顯的提升效果.

3.2 算法的并行化效果和吞吐量

3.2.1 實驗數據

測試數據源為UCI數據倉庫中的數據集Nursery,數據集屬性集描述如表4所示：

Table 4 Attributes Description of Nursery Dataset

利用Kafka-2.8.0產生消息隊列模擬實時數據流，分別建立訓練數據流和測試數據流2個topic，供建樹Topology和分類拓撲中的KafkaSpout讀取.使用storm jar命令將封裝好的P-HT算法jar包提交到Storm集群上運行拓撲，這里設置P-HT算法的并行度為4.利用Storm UI觀察算法運行的線程壓力和吞吐量.

3.2.2 實驗環境

1) 本地模式

CPU 2.13 GHz、3.11 GB內存；

Eclipse Release 4.2.0；

JRE1.7.05_25；

Redis-2.4.5.

2) 集群模式

① 軟件環境

Storm版本：apache-storm-0.9.1；

Kafka版本：2.8.0-0.8.1.1；

Redis版本：Redis-2.8.13 64位；

Java版本：JDK1.7.0_55；

Zookeeper 版本：Zookeeper-3.4.6.

② 硬件環境

Storm集群由master，node1～node4等5個物理節點組成，分別負責Storm中拓撲程序的控制和計算工作，其中控制節點為master，上面運行Storm的Nimbus進程，計算節點為node1～node4，上面運行Storm的Supervisor后臺進程及算法運算的Worker進程.

3.2.3 算法參數配置

1) 當幾個屬性的信息熵G相同或者差別很小，引入ties值，主動選擇是否可以分裂形成新的子樹，tieConfidence=0.05.

2) 用于計算Hoeffding bounds條件的可信度δ=1E-4.

3) 初始滑動窗口大小windowsize=200 000.

4) 檢測節點分裂有效性間隔樣本數為20 000.

5) 檢測概念漂移間隔為1 000.

6) 檢測概念漂移時采樣的樣本數為1 000.

3.2.4 實驗結果

1) 并行化效果分析

P-HT算法分類拓撲在進行分類時是利用從Redis中讀取的實時更新的分類器，將Kafka發送的待分類數據分配給多個線程進行分布式地打標簽，并且在分類結果處理Bolt將合并統計所有Map線程分類的結果.直接測量算法每秒處理的數據量是不準確的，所以本節將控制流入拓撲的數據流速，并且相應改變P-HT算法的并行度，通過Storm的UI界面觀測各線程的線程處理壓力capacity(capacity等于Bolt調用excute方法處理的消息數量乘以消息的平均時間除以時間區間).我們控制每秒KafkaSpout發送待分類數據樣本20 000條.分別修改分類Bolt的線程數為1，2，4時各線程的處理壓力，并行度分別為1，2，4時線程capacity的測試結果見圖7所示：

Fig. 7 Relation graph of thread capacity and parallelism圖7 線程capacity與并行度關系圖

由測試結果可以看出，單個Map線程的處理壓力隨著并行度的增加，呈倒數減小趨勢；而Reduce線程的壓力隨著Map線程的增加，呈近線性增加趨勢.該測試結果與理論分析一致：n個線程分布式計算與單機模式相比，處理相同數量的數據，單個Map線程的處理壓力約降為1n；而由于每個分布式線程發送相同數目的分類結果供合并線程合并，Reduce線程的壓力隨著并行度的提高呈線性增加趨勢，但Map線程和Reduce線程的壓力均始終維持在0.4以下，因此垂直并行化的方法保證了算法的效率.

2) 抗概念漂移性能分析

由于流數據環境下，數據的分布式不穩定，會隨時出現概念漂移的情況，本文采用Nursery數據集模擬發生概念漂移的數據流，對VFDT算法、CVFDT算法和P-HT算法分別進行準確率的測試，實驗結果如圖8所示.

Fig. 8 Performance analysis of anti concept drift of P-HT algorithm圖8 P-HT算法抗概念漂移性能分析

由圖8可知，數據量達到40 000條之前，3種算法的準確率差別不大，在數據量達到40 000條時由于出現了概念漂移，3種算法的準確率均出現了不同程度的下降，P-HT算法的準確率突降的幅度是3種算法中最小的.由于采用了替代子樹檢測概念漂移機制，CVFDT算法準確率下降的幅度要小于VFDT算法，由于P-HT算法引入了可變的滑動窗口機制，其準確率不僅下降的幅度最小，同時在準確率突降之后數據流恢復平穩時，P-HT算法準確率的提升速度也是3種算法中最快的.

3) 吞吐量分析

本文基于決策樹的垂直并行化思想對于串行建樹的流分類CVFDT算法進行了改進，在決策樹內部節點的分裂過程中，利用Storm集群的特點并行的計算分裂節點的最佳分裂屬性，實驗將CVFDT算法和P-HT算法拓撲提交到Storm集群上，利用相同的Nursery數據集產生數據流，對于2個算法處理的數據量進行統計，實驗結果如圖9所示：

Fig. 9 Comparison of parallel P-HT and serial CVFDT圖9 并行P-HT與串行CVFDT吞吐量對比

從圖9可以看出，在算法的初始化階段，由于數據量沒有達到決策樹的生長所需的最小樣本數，P-HT算法和CVFDT算法都處于樣本流進入窗口階段，因此串行CVFDT算法和并行化P-HT算法的處理數據速度并沒有較大的差別；但是隨著算法拓撲在Storm集群中運行時間的增加，數據量對算法的速率要求開始提高，可以看出并行化P-HT算法所處理的數據量要明顯大于串行處理方式下的CVFDT算法，體現了本文提出的垂直并行化方法對于流數據分類算法的速率有顯著的效果.

4 結束語

本文介紹了傳統分類算法和Storm平臺的特點，在深入研究流分類VFDT算法和Storm平臺的基礎上，結合Storm實時流處理平臺的天然優勢，提出基于Storm平臺的流分類P-HT并行化算法.該算法引入了時間滑動窗口模型，保證了分類模型的實時性和準確率.同時，算法結合了Storm集群的分布式的快速、高效特點，實現了傳統決策樹算法的并行化建樹，提高了算法的訓練速度和處理效率，使得傳統的決策樹分類算法在大數據流環境下得到應用.實驗結果表明，基于Storm的P-HT算法在擁有和離線分類挖掘相當的準確率的同時，比串行的流分類算法擁有更大的處理速度和吞吐量.

[1]Suzuki Y, Kido K. Big-data streaming applications scheduling with online learning and concept drift detection[C]Proc of Design, Automation & Test in Europe. Piscataway, NJ: IEEE, 2015: 1547-1550

[2]Wang Tao, Li Zhoujun, Yan Yuejin, et al. A survey of classification of data streams[J]. Journal of Computer Research and Development, 2007, 44(11): 1809-1815 (in Chinese)(王濤, 李舟軍, 顏躍進, 等. 數據流挖掘分類技術綜述[J]. 計算機研究與發展, 2007, 44(11): 1809-1815)

[3]Gaber M M, Zaslavsky A, Krishnaswamy S. Data Stream Mining[G]Data Mining and Knowledge Discovery Handbook. Berlin: Springer, 2009: 759-787

[4]Quinlan J. Induction of decision trees[J]. Machine Learning, 1986, 1(1): 81-106

[5]Street W N, Kim Y S. A streaming ensemble algorithm (SEA) for large-scale classification[C]Proc of the 7th ACM SIGKDD Int Conf on Knowledge Discovery and Data Mining. New York: ACM, 2001: 377-382

[6]Domingos P, Hulten G. Mining high-speed data streams[C]Proc of the 6th ACM SIGKDD Int Conf on Knowledge Discovery and Data Mining. New York: ACM, 2002: 71-80

[7]Gama J, Rocha R, Medas P. Accurate decision trees for mining high-speed data streams[C]Proc of the 9th ACM SIGKDD Int Conf on Knowledge Discovery and Data Mining. New York: ACM, 2003: 523-528

[8]Gama J, Fernandes R, Rocha R. Decision trees for mining data streams[J]. Intelligent Data Analysis, 2006, 10(1): 23-45

[9]Jin Ruoming. Efficient decision tree construction on streaming data[C]Proc of the 9th ACM SIGKDD Int Conf on Knowledge Discovery and Data Mining. New York: ACM, 2003: 571-576

[10]Anagnostopoulos C, Tasoulis D K, Adams N M, et al. Temporally adaptive estimation of logistic classifiers on data streams[J]. Advances in Data Analysis & Classification, 2009, 3(3): 243-261

[11]Kuncheva L I. Classifier ensembles for changing environments[G]Multiple Classifier Systems. Berlin: Springer, 2004: 1-15

[12]Gama J. A survey on learning from data streams: Current and future trends[J]. Progress in Artificial Intelligence, 2012, 1(1): 45-55

[13]Hulten G, Spencer L, Domingos P. Mining time-changing data streams[C]Proc of the ACM SIGKDD Int Conf on Knowledge Discovery and Data Mining. New York: ACM, 2001: 97-106

[15]Liang Chunquan, Zhang Yang, Shi Peng, et al. Learning very fast decision tree from uncertain data streams with positive and unlabeled samples[J]. Information Sciences, 2012, 213(23): 50-67

[16]Zheng Wenhua. Constructing decision trees for mining high-speed data streams[J]. Chinese Journal of Electronics, 2012, 21(2): 215-220

[17]Quinlan J R. Improved use of continuous attributes in C4.5[J]. Journal of Artificial Intelligence Research, 1996, 4(1): 77-90

[18]Rahnama A H A. Distributed real-time sentiment analysis for big data social streams[C]Proc of Int Conf on Control, Decision and Information Technologies (CoDIT). Piscataway, NJ: IEEE, 2014: 789-794

Ji Yimu, born in 1978. Professor. His main research interests include P2P network optimization, cloud computing security and stream data query in big data.

Zhang Yongpan, born in 1994. Master. His current research interests include stream query and mining in big data.

Lang Xianbo, born in 1991. Master. His current research interest is stream query of join in big data.

Zhang Dianchao，born in 1990. Master. His main research interests include the application of big data platform architecture, and computing security.

Wang Ruchuan, born in 1943. Professor. His main research interests include IoT, cloud computing and big data.

《計算機研究與發展》征訂啟事

《計算機研究與發展》(Journal of Computer Research and Development)是中國科學院計算技術研究所和中國計算機學會聯合主辦、科學出版社出版的學術性刊物，中國計算機學會會刊.主要刊登計算機科學技術領域高水平的學術論文、最新科研成果和重大應用成果.讀者對象為從事計算機研究與開發的研究人員、工程技術人員、各大專院校計算機相關專業的師生以及高新企業研發人員等.

《計算機研究與發展》于1958年創刊，是我國第一個計算機刊物，現已成為我國計算機領域權威性的學術期刊之一.并歷次被評為我國計算機類核心期刊，多次被評為“中國百種杰出學術期刊”.此外，還被《中國學術期刊文摘》、《中國科學引文索引》、“中國科學引文數據庫”、“中國科技論文統計源數據庫”、美國工程索引(Ei)檢索系統、日本《科學技術文獻速報》、俄羅斯《文摘雜志》、英國《科學文摘》(SA)等國內外重要檢索機構收錄.

國內郵發代號：2-654；國外發行代號：M603

國際標準連續出版物號：ISSN1000-1239

聯系方式：

100190 北京中關村科學院南路6號《計算機研究與發展》編輯部

電話： +86(10)62620696(兼傳真)；+86(10)62600350

Email:crad@ict.ac.cn

Parallel of Decision Tree Classification Algorithm for Stream Data

Ji Yimu1,2,3,4, Zhang Yongpan1, Lang Xianbo1, Zhang Dianchao1, and Wang Ruchuan1,2

1(SchoolofComputerScienceandTechnology,NanjingUniversityofPostsandTelecommunications,Nanjing210023)2(JiangsuHighTechnologyResearchKeyLaboratoryforWirelessSensorNetworks(NanjingUniversityofPostsandTelecommunications),Nanjing210023)3(InstituteofAdvancedTechnology,NanjingUniversityofPostsandTelecommunications,Nanjing210023)4(KeyLaboratoryofIntelligentPerceptionandSystemsforHigh-DimensionalInformation(NanjingUniversityofScienceandTechnology),MinistryofEducation,Nanjing210094)

With the rise of cloud computing, Internet of things and other technologies, streaming data exists widely in telecommunications, Internet, finance and other fields as a new form of big data. Compared with the traditional static data, stream data in big data has the characters of rapidness, continuity and changing with time. At the same time, the implicit distribution of the data stream will bring about the concept drift problem. In order to satisfy the requirements of stream data classification algorithms in big data, we must improve the traditional static offline data classification algorithms, and propose P-HT parallel algorithm based on distributed computing platform Storm. To meet the requirements of Storm stream processing platform, we improve the flexibility and versatility of the algorithm through sliding window mechanism, alternative tree mechanism and parallel processing mechanism, and the algorithm can adapt to the concept-drift of data stream very well. Finally, we experimentally verify the validity and high efficiency of the algorithm. The results show that the improved P-HT algorithm has better throughput and faster processing speed than the traditional C4.5 algorithm in the case of no reduction in accuracy.

stream data; classification algorithms; Storm platform; sliding windows; C4.5 algorithm; paralleling algorithm

2016-08-02；

2016-12-09

國家自然科學基金項目(61170065)；江蘇省自然科學基金優秀青年基金項目(BK20170100)；國家重點研發計劃(2017YFB0202200);江蘇省重點研發計劃項目(BE2017166) This work was supported by the National Natural Science Foundation of China (61170065), Outstanding Youth of Jiangsu Natural Science Foundation (BK20170100), the National Key Research and Development Program of China (2017YFB0202200), and the Jiangsu Key Research and Development Program (BE2017166).

TP391