一種改進的并行關聯規則增量更新算法研究

王 誠，趙申屹

(南京郵電大學 通信與信息工程學院，江蘇 南京 210003)

0 引 言

在現實的挖掘過程中通常存在增量更新[1]問題。挖掘對象的數據集和支持度會發生變化，使用靜態關聯挖掘方法需要反復對更新后的數據集進行掃描，原有的挖掘結果失去作用，在面對大規模數據集時挖掘效率低下。將并行化計算框架與增量關聯算法相結合，不僅能夠提高挖掘效率，而且在海量數據挖掘中有實際意義。

增量關聯規則[2]出現至今，已有了眾多的研究成果。Agrawal提出了關聯規則挖掘中最著名的Apriori算法。如今的大部分增量關聯規則算法都是基于Apriori算法的改進或擴展。Cheung等闡述了數據集變大的增量關聯規則挖掘問題，提出了在數據集增加情況下的FUP算法[3]。該算法通過比較原始事務數據庫和新增數據庫的項集之間的頻繁和非頻繁關系，對頻繁項集進行增量更新得到更新后事務數據庫的頻繁項集。由于FUP是基于Apriori算法產生的，需要多次掃描原始數據集并且生成大量候選項集，該算法仍存在挖掘效率低下的問題。文獻[4]基于FUP算法，提出了改進的UWEP算法。文獻[5]將FUP算法思想應用到FP-Growth算法中，提出了FUFP-tree算法。該算法在數據更新時，只需掃描原始事務數據庫中的變動部分，無須掃描整個數據庫，從而大幅提高了挖掘效率。

針對傳統單機關聯挖掘方法在海量數據環境下挖掘效率低的問題，引入分布式并行處理框架[6-8]，如Hadoop、Storm、Spark等。文獻[9]在FUFP-tree算法基礎上，提出了一種基于MapReduce編程框架的并行化模式算法-PFUFP-tree。文獻[10]結合MapReduce和FUP算法提出了一種并行的增量關聯規則算法-MRFUP。文獻[11]在Apriori算法的基礎上，提出了一種基于Spark框架的并行化算法-AMRDD。

Spark[12]作為一個新興的大數據處理引擎，允許用戶將數據加載至內存后重復使用。其基于內存的計算特性使其大數據計算性能大大超過MapReduce。針對海量數據下的關聯規則挖掘和增量更新問題，基于PFP-tree算法[13]的思想，提出一種改進的關聯規則增量更新算法(parallel updated frequent pattern growth algorithm on Spark，SPUFP)。該算法優化了頻繁模式樹結構和并行計算分組策略[14]，減小了時間和空間復雜度，并結合Spark編程框架[15]進行實現，使其能夠高效運行于Spark平臺，大幅提高了海量數據環境下的挖掘效率。

1 增量更新算法

增量關聯規則挖掘是指數據集變化或者支持度變化時的關聯規則挖掘。Cheung等提出的快速更新算法FUP，是基于Apriori的增量更新算法，把增量更新后的項集分成4種類型，針對數據集增大的情況進行研究。FUP算法的第k次循環僅需掃描數據庫一次，候選項集生成新的頻繁項集前會根據在d上的支持度先進行修剪，挖掘效率相比更新后的數據集中直接使用Apriori算法高很多。但是在對候選項集支持度進行比較時，該算法仍需多次重復掃描數據集來找出所有的頻繁項集，會因計算量的增大導致計算速度緩慢，內存消耗過大。

PFP-tree算法是基于FP-Growth的并行算法，在數據輸入前，將原始數據集劃分且存儲到不同的節點上，通過掃描數據庫，能夠并行地計算各個節點上項的支持度計數，產生頻繁1項集，降序排序生成FList。之后，將FList分成多個小組，每組包含若干個項，組成GList，再對每個組的事務組按照傳統的FP-Growth創建FP-tree，根據GList中的項對FP-tree進行遞歸的頻繁挖掘。然而PFP-tree算法在更新樹結構的過程中需要把更新后的數據集壓縮到頻繁模式樹中，會消耗大量的時間和存儲空間，當數據增量較大時，樹的規模會非常大，導致挖掘效率低下。其次，PFP-tree算法在FList的分組步驟中，將FList根據并行計算的節點數平均分為g個組，沒有充分考慮不同事務組計算的負載量不同的問題。

2 改進的關聯規則增量更新算法

SPUFP算法的主要思想如下：在增量更新過程中，利用已挖掘的原始數據集DB的中間結果建立更新后數據集S的頭表，僅需要將新增數據集db壓縮到頻繁模式樹中，排除了大量在db中頻繁而在S中非頻繁的項集，減少了挖掘時間，大幅減小了樹的規模，釋放出大量內存空間。針對并行計算中的分組問題，提出了一種優化的分組策略，令靠前的分組對更多的項進行挖掘，從而實現分組間的負載均衡。

2.1 算法描述

設定原始事務數據庫為DB，事務集T={T1,T2,…,Tn}，候選項集為CD，頻繁項集為LD，原始事務數據庫大小為|DB|；新增數據庫為db，頻繁項集為Ld，新增數據庫大小為|db|；更新后的數據庫為S，S=DB∪db，頻繁項集為LS，支持度為sup。

輸入：原始事務數據集DB；新增事務數據集db；

輸出：更新后的數據集S的頻繁項集LS。

步驟1：掃描原始數據集，得到DB的候選1項集CD1，根據支持度得出頻繁1項集LD1，排序后建立FList，按照分組策略進行分組，構造DB的頻繁模式樹，對頻繁模式樹進行頻繁項集的挖掘，得到DB的頻繁項集LD。

步驟2：掃描新增數據集，得到db的候選1項集Cd1，讀取步驟1中的CD1，合并得到更新后數據集S的候選1項集CS1，根據支持度得出S的頻繁1項集SD1，排序后建立FList’，按照分組策略進行分組，構造db的頻繁模式樹。

步驟3：對頻繁模式樹進行頻繁項集的挖掘，同時讀取步驟1中DB的頻繁項集LD，對生成的每個頻繁模式k和LD做比較：若k屬于LD，則k是原數據集的頻繁項，將k的支持度與在LD對應的支持度計數相加可得k在S中的支持度計數，如果總計數大于S的最小支持度計數，則加入S的部分頻繁項集L'，并從LD中刪除該項；若k不屬于LD，則k是新增數據集的頻繁項，不能確定在S中是否頻繁，將其加入S的候選集CS。判別LD中剩余的項，如果其支持度計數大于S的最小支持度計數，則加入L'。

步驟4：掃描原始數據集，讀取步驟3中S的候選集CS，判斷是否為S的頻繁項集，將頻繁項集與部分頻繁項集L'合并，便得到更新后數據集S的頻繁項集LS。

2.2 分組策略

在利用頻繁1項集排序建立頭表FList后，為執行并行計算，需要將FList中的項目分成g個小組，把每個組分得的項目存入名為GList的表中，再分發至各個節點展開并行計算。文中分組方法如下：設定FList中的項目數量為M，GList中小組數量為g，指針FS和FE分別指向FList的表頭和末尾，指針GS和GE分別指向GList的表頭和末尾。

(1)如果2g≤M，則將FList中FS到FS+M/2全部的項目分到GList中的GS，FS指向FS+M/2+1，GS指向GS+1，令M=M/2，g=g-1，轉向步驟4；

(2)如果2g>M，則將FList中FE指向的項目分到GList中的GE，FE指向FE-1，GE指向GE+1，令M=M-1，g=g-1，轉向步驟4；

(3)如果g>1，返回步驟1、2；

(4)如果g=1，則將FList中剩余的項目全部分到GList剩余的最后一組中，分組結束。

3 基于Spark并行計算實現

3.1 Spark編程模型

針對傳統的增量關聯規則算法在面對海量數據集時挖掘效率低的問題，借助Spark并行計算框架對算法進行改進，提高增量更新效率。

Spark是一個通用的大規模數據快速處理引擎，將分布式數據抽象為彈性分布式數據集RDD。RDD是Spark計算的基礎，支持并行操作，允許用戶將數據加載至內存中重復地使用，一個RDD代表一個分區里的數據集。每個RDD的數據都以Block的形式存儲于多臺機器上。Spark的RDD存儲架構如圖1所示。

圖1 RDD存儲架構

Spark支持兩種RDD的基本操作(見表1)：轉換Transformation和動作Action。轉換操作屬于延遲計算，將一個數據集轉換成新的數據集，當一個RDD轉換成另一個RDD時并沒有立即進行轉換，僅僅是記住了數據集的邏輯操作。動作操作對數據集進行計算并將計算結果返回給驅動程序，觸發Spark作業的運行，真正觸發轉換算子的計算。Spark中基于RDD的所有轉換并不會即刻被執行，而是直到出現動作操作請求返回結果時，轉換操作才被執行，減少了不必要的計算和返回。另外，RDD支持內容的持久化，把內容保存在各節點的內存中，不會被覆蓋或者刪除，這樣下次調用RDD時就不必創建新的RDD，加快了計算速度。

表1 RDD基本操作

Spark的結構與MapReduce類似，由一個Master節點和若干個worker組成。用戶通過編寫程序Driver與Master進行交互，將所有定義好的RDD操作提交到Master，Master再把接受的RDD操作分發給各個worker。worker收到操作后，根據數據分塊的信息，選擇相應數據進行操作，生成新的RDD。

3．2 基于Spark的增量頻繁模式樹算法

文中的關聯增量更新算法基于Spark并行計算框架實現，算法主要分為兩個階段。第一階段，對原始數據集DB進行并行挖掘并保留計算結果；第二階段，在新增數據庫db中，通過使用原數據庫DB的挖掘結果完成更新后數據庫S的頻繁項挖掘。

3.2.1 對原始數據庫的挖掘

設定原始事務數據庫為DB，事務集T={T1,T2,…,Tn}，候選項集為CD，頻繁項集為LD，原始事務數據庫大小為|DB|；新增數據庫為db，頻繁項集為Ld，新增數據庫大小為|db|；更新后的數據庫為S，S=DB∪db，頻繁項集為LS，支持度為sup。

輸入：原始數據集DB；

輸出：原始數據集DB的頻繁項集LD。

步驟1：將原始數據集DB存儲到分布式文件系統HDFS中，數據集會被分割成若干個數據塊，分發到各個工作節點上，每個數據分塊都分別由多個事務Ti={item1,item2,…,itemn}組成。通過textfile讀取數據并存入RDD中，對每個數據分塊進行flatmap操作，過程中遍歷所有事務Ti中的項，以itemi為鍵，1為值，輸出(item,1)形式的鍵值對。再利用reduceByKey操作累計項目數，將擁有相同鍵的值進行累加，輸出(item, count)的鍵值對，count表示對應項的總計數，得到原始數據集DB的候選1項集CD1。

步驟2：以步驟1的結果作為輸入，通過filter操作，過濾掉總計數低于最小支持度的項，剩下的項即為頻繁1項集LD1。讀取LD1，把所有項根據計數count降序排列存入FList，再根據并行計算的節點數量分成m個組，得到GList。通過broadcast操作將GList轉換成全局共享變量，以緩存形式分發到各個節點。

步驟3：GList以哈希表的形式存儲，以項為鍵，項所在的分組號g為值。讀取每個事務Ti，然后對事務中的每個項item取出Glist中的分組號g，輸出(g，(item1,item2,…，itemn))的鍵值對。通過groupByKey操作，將擁有相同分組號的鍵值對合并，發送到同一個工作節點，輸出得到(g,Dg)，Dg為各分組號對應的事務組。利用foreach操作，對每個分組g，讀取其在GList中對應的部分，掃描事務組Dg，創建根節點為null的FP-tree。最后調用growth方法，掃描各個節點對應的GList部分，輸出以pattern為鍵，支持度為值的鍵值對，得到原始數據集的頻繁項集LD，保存到HDFS中。

3.2.2 增量更新

輸入：原始數據集DB的頻繁項集LD；新增數據集db；

輸出：更新后數據集S的頻繁項集LS。

步驟1：通過textfile將新增數據集db構成RDD，db同樣被分布在各個工作節點中。類似3.1中的步驟，通過flatmap操作遍歷事務中所有項得到(item,1)形式的鍵值對，再通過reduceByKey操作累計項目數得到(item, count)的鍵值對，即為新增數據集db的候選1項集Cd1。讀取DB的候選1項集CD1，將結果與之合并，得到更新后數據集S的候選1項集CS1。

步驟2：以S的候選1項集CS1作為輸入，通過filter操作，過濾掉總計數低于最小支持度的項，剩下的項即為S的頻繁1項集SD1，按支持度計數降序排序后生成FList’，按計算節點數分組得到GList’。讀取DB的分組GList，將在DB中非頻繁而更新后頻繁的項加入分組；將在DB中頻繁而更新后非頻繁的項從分組中刪除，得到處理后的GList*。通過broadcast操作將GList*轉換成全局共享變量。通過foreach，對每個分組g，讀取原始數據集DB在該分組的頻繁項集LD，掃描新增數據集在分組中的事務組dg，創建FP-tree。調用growth方法進行挖掘，過程中掃描DB相應分組的頻繁項集LD，對產生的每個鍵值對k，若k屬于LD，則k是原數據集的頻繁項，將k的支持度與在LD對應的支持度計數相加可得k在S中的支持度計數，通過filter操作過濾掉總計數低于最小支持度的項，剩下的項為S的部分頻繁項集L'；若k不屬于LD，則k是新增數據集的頻繁項，不能確定在S中是否頻繁，將其加入S的候選集CS。

步驟3：讀取原始數據集DB，對每個數據分塊，根據GList*將DB發送到相應的分組中。對于每個分組g，讀取該分組的候選項集CS，掃描原始數據集DB在該分組的事務組，根據最終生成的支持度計數，與最小支持度進行比較，得到候選項集CS中包含的S的頻繁項集。將結果與步驟2中的部分頻繁項集L'相加，即為更新后數據集S的頻繁項集LS。

4 實 驗

實驗所用主機為Intel(R)Core(TM)i3,主頻2 GHz，內存2 GB，操作系統為Ubuntu16.04 32位,1臺為Master節點，4臺為worker節點，每個節點的配置差異很小。軟件環境為Hadoop2.7.0，JDK1.8，Spark2.1.0，scala2.12。

實驗數據來自http://fimi.ua.ac.be/data/的webdocs.dat.gz數據集，容量約為1 450 MB。

4.1 單機環境下的算法性能分析

實驗僅在Master節點上將數據集平均分成大小相同的10組，每組包括145 MB數據。分別將2組、4組、6組、8組數據作為原始數據集DB，標記為D1、D2、D3、D4,2組數據作為更新數據集db，最小支持度設為10%，對SPUFP算法和PFP-tree算法、傳統FUP算法的運行時間進行比較，如圖2所示。

圖2 不同算法運行時間比較

從圖2可以看出，當數據量較小時，基于并行計算框架的算法和傳統的單機算法的耗時差別不大。隨著更新數據集的增加，兩者計算的時間差開始增大，并行算法的挖掘效率明顯提高且保持穩定。當更新數據量繼續增長后，SPUFP算法比基于MapReduce[16-17]的PFP-tree算法有了較大的優勢。這是因為在處理數據量較小時，并行算法調度操作的時間占用算法運行總時間的比例較大；當更新數據量逐步增大，此比例開始減小，并行算法效率開始優于傳統算法；而隨著數據集進一步增大，由于Spark能夠將計算的中間結果緩存在內存中，節省了大量I/O操作消耗的時間，所以效率較MapReduce算法又有了明顯提升。

4.2 分布式集群環境下的算法性能分析

用多個參數相同的節點搭建Spark集群環境測試算法的可擴展性，將數據集平均分成大小相同的10組，每組包括145 MB數據。分別將2組、4組、6組、8組數據作為原始數據集DB，標記為D1、D2、D3、D4，2組數據作為更新數據集db，最小支持度設置為10%。分別使用1到4個worker節點測試SPUP算法的運行時間，如圖3所示。

圖3 不同節點數運行時間比較

從圖3可以看出，在數據量相同條件下，文中并行算法的運行時間隨著worker節點數量的增加不斷減少，且更新后的數據總量越大，時間減少的幅度越大，說明SPUFP算法在數據集較大的環境下有良好的可擴展性。另一方面，隨著節點個數的增加，不同容量的數據集的運行時間差別逐漸變小，這是因為節點增加會導致集群內節點間通信開銷增大。

5 結束語

基于頻繁模式樹的思想，提出了一種改進的并行關聯規則增量更新算法，優化了頻繁模式樹結構和并行計算分組策略，有效減少了挖掘時間和存儲空間。實驗結果表明，Spark靈活的并行計算框架具備良好的可擴展性，其基于內存的計算特點在大規模數據環境下效率高于Hadoop。在今后的研究中，可以通過Spark等云計算平臺對更多傳統數據挖掘算法進行改進，提高海量數據下的挖掘效率。