一種基于Spark的高效增量頻繁模式挖掘算法

荀亞玲，孫嬌嬌，畢慧敏

(太原科技大學(xué)計(jì)算機(jī)科學(xué)與技術(shù)學(xué)院，太原030024)

頻繁項(xiàng)集挖掘(FIM)是一項(xiàng)非常重要的數(shù)據(jù)挖掘任務(wù)，其主要任務(wù)是查找交易數(shù)據(jù)庫(kù)中經(jīng)常一起出現(xiàn)的項(xiàng)目集、序列或子結(jié)構(gòu)[1]。 FIM已經(jīng)引起了數(shù)據(jù)挖掘領(lǐng)域和各種實(shí)際應(yīng)用領(lǐng)域的廣泛關(guān)注[2-3]。但是，其挖掘過(guò)程占用大量CPU和I/O，因?yàn)闀?huì)生成大量中間項(xiàng)和結(jié)果項(xiàng)集，并且此過(guò)程中涉及的數(shù)據(jù)集也很大。

本文采用一種無(wú)損樹(shù)結(jié)構(gòu)(FCFP-Tree)[4]，通過(guò)在維護(hù)與新增數(shù)據(jù)相對(duì)應(yīng)的樹(shù)結(jié)構(gòu)時(shí)避免重新掃描原始數(shù)據(jù)集，以顯著減少I(mǎi)/O.為了突破單機(jī)資源進(jìn)行大數(shù)據(jù)處理的限制，開(kāi)發(fā)了一種基于內(nèi)存計(jì)算Spark平臺(tái)的分布式解決方案，該解決方案設(shè)計(jì)了更為合理的RDD轉(zhuǎn)換策略并采用相關(guān)組投影技術(shù)來(lái)優(yōu)化通信和計(jì)算開(kāi)銷(xiāo)。

1 相關(guān)研究

關(guān)聯(lián)規(guī)則挖掘一直以來(lái)都是數(shù)據(jù)挖掘研究的熱點(diǎn)課題，由于數(shù)據(jù)集往往隨著時(shí)間的變化而不斷積累，因此，原先的挖掘結(jié)果可能已不適用，為了有效挖掘新的模式，增量挖掘算法被廣泛研究。增量挖掘算法大致可分為三類(lèi)：(1)基于Apriori的增量挖掘：如早期的FUP算法和相應(yīng)的改進(jìn)算法FUP2算法[5]，它們?nèi)岳^承了Apriori的固有缺點(diǎn)，即需要多次掃描原始數(shù)據(jù)集；TDUP算法亦是基于Apriori的增量挖掘算法[6]，先利用Apriori算法從初始數(shù)據(jù)集中提取正區(qū)域和邊界區(qū)域中的項(xiàng)目集，而后根據(jù)增量數(shù)據(jù)更新項(xiàng)目集的支持計(jì)數(shù)，僅考慮和計(jì)算邊界區(qū)域中的那些項(xiàng)集，由于不考慮負(fù)區(qū)域的項(xiàng)目集，因此TDUP可能會(huì)錯(cuò)過(guò)一些具有增量數(shù)據(jù)更新的潛在頻繁項(xiàng)目集；(2)基于分區(qū)的增量挖掘：為了避免類(lèi)-Apriori算法需多次掃描數(shù)據(jù)庫(kù)并生成大量候選集的問(wèn)題，一種基于分區(qū)的滑動(dòng)窗口過(guò)濾(SWF)算法被提出[7]，其基本思想是將事務(wù)數(shù)據(jù)庫(kù)被分成n個(gè)分區(qū)，并在每個(gè)分區(qū)中采用過(guò)濾閾值來(lái)處理候選項(xiàng)集生成；其后產(chǎn)生了一些列基于SWF的改進(jìn)算法，Li H.F.等人提出一種有效的基于bit的挖掘算法，該算法由三個(gè)階段組成[8]，第一個(gè)是窗口初始化，每個(gè)事務(wù)中的每個(gè)項(xiàng)目被編碼為有效的比特序列表示；其次，左移技術(shù)用于在窗口滑動(dòng)階段期間有效地滑動(dòng)窗口；最后，在模式生成階段期間水平生成當(dāng)前滑動(dòng)窗口中的完整頻繁項(xiàng)目集；(3)基于樹(shù)結(jié)構(gòu)的增量挖掘：以FP-Tree為代表的樹(shù)結(jié)構(gòu)大大提高了頻繁項(xiàng)集挖掘效率。Koh J.L.等提出了AFPIM算法[9]，該算法仍然采用FP-Tree存儲(chǔ)數(shù)據(jù)，但其設(shè)置了一個(gè)低于最小支持度閾值的參數(shù)preMinsup，在樹(shù)中保留支持度大于preMinsup的項(xiàng)目才會(huì)保留在樹(shù)中，一定程度上避免增量挖掘時(shí)對(duì)數(shù)據(jù)庫(kù)的重復(fù)掃描，但當(dāng)先前“不頻繁”的項(xiàng)在更新的數(shù)據(jù)集中變得“頻繁”時(shí)，AFPIM算法不能僅通過(guò)調(diào)整原始樹(shù)中的項(xiàng)目來(lái)生成更新的FP-Tree，它需要重新掃描整個(gè)更新的數(shù)據(jù)集以構(gòu)建新的FP-Tree；為了避免樹(shù)的重新構(gòu)建，CATS-Tree[10]和Can-Tree[11]相繼被提出，但是他們的壓縮度又不夠。

如今，大數(shù)據(jù)時(shí)代的到來(lái)，傳統(tǒng)串行算法在計(jì)算效率、可擴(kuò)展等方面均不能滿足需求。Li N.等人提出基于分布式處理框架的PApriori算法[12]，其是標(biāo)準(zhǔn)Apriori的簡(jiǎn)單分布式實(shí)現(xiàn)，但是仍需要多次掃描數(shù)據(jù)庫(kù)；基于FP-Tree和Can-Tree的并行實(shí)現(xiàn)算法IncMiningPFP和IncBuildingPFP被提出[13]；上述基于Hadoop MapReduce的并行算法雖在性能上取得不同程度的改進(jìn)，然而通過(guò)磁盤(pán)進(jìn)行大量讀寫(xiě)操作極大地限制了它們的性能。Qiu O.H.等人提出了YAFIM算法[14]，該算法充分利用了Spark框架的RDD編程和基于內(nèi)存的并行計(jì)算等特點(diǎn)，大大提高了挖掘效率；Yang S.等人引入一種新的基于矩陣的剪枝方法[15]，減少候選項(xiàng)集的數(shù)量，避免不必要的搜索開(kāi)銷(xiāo)；Sethi等人利用垂直布局解決了每次迭代過(guò)程中掃描數(shù)據(jù)集的問(wèn)題[16]。

綜上所述，目前大多數(shù)算法，在進(jìn)行增量挖掘時(shí)不能有效利用原始挖掘結(jié)果，仍需重新掃描原始數(shù)據(jù)庫(kù)，以及重新構(gòu)建樹(shù)造成極大的資源浪費(fèi)；而并行算法雖然使得挖掘性能得到了進(jìn)一步提升，但基于Apriori的多次數(shù)據(jù)庫(kù)數(shù)據(jù)庫(kù)掃描和基于樹(shù)結(jié)構(gòu)的巨大樹(shù)結(jié)構(gòu)調(diào)整代價(jià)，成為了目前并行算法的瓶頸。本文結(jié)合一種新的樹(shù)結(jié)構(gòu)，避免了原始數(shù)據(jù)庫(kù)的掃描和高昂的樹(shù)結(jié)構(gòu)調(diào)整代價(jià)，并通過(guò)設(shè)計(jì)有效的RDD轉(zhuǎn)換策略和負(fù)載均衡優(yōu)化方案實(shí)現(xiàn)高效的并行頻繁模式挖掘。

2 全壓縮頻繁模式樹(shù)及樹(shù)結(jié)構(gòu)調(diào)整策略

2.1 全壓縮頻繁模式樹(shù)(FCFP-Tree)

在增量挖掘過(guò)程中，為了避免重新掃描原始數(shù)據(jù)集和重構(gòu)樹(shù)結(jié)構(gòu)的開(kāi)銷(xiāo)，本文采用FCFP樹(shù)結(jié)構(gòu)壓縮和存儲(chǔ)原始數(shù)據(jù)集。

定義1FCFP-Tree：FCFP-Tree是一種樹(shù)結(jié)構(gòu)，F(xiàn)CFP樹(shù)由四部分組成。第一部分是根節(jié)點(diǎn)，標(biāo)記為“null”.第二部分是所有子項(xiàng)的樹(shù)，項(xiàng)目前綴子樹(shù)為根。第三部分是用于存儲(chǔ)頻繁項(xiàng)目的頭表。第四部分是用于存儲(chǔ)不頻繁項(xiàng)目的頭表。

項(xiàng)前綴子樹(shù)中的每個(gè)節(jié)點(diǎn)由四個(gè)域組成，F(xiàn)CFP樹(shù)的節(jié)點(diǎn)結(jié)構(gòu)定義如下：

Struct FCFPNode {

Int NodeCount；

String NodeItem；

Struct FCFPNode * NodeParent；

Struct FCFPNode * NodeLink；

}

NodeItem：節(jié)點(diǎn)的名稱(chēng)，可以標(biāo)識(shí)項(xiàng)目。NodeCount：節(jié)點(diǎn)的統(tǒng)計(jì)計(jì)數(shù)，是NodeItem標(biāo)識(shí)的項(xiàng)的相應(yīng)支持計(jì)數(shù)。NodeParent：指向FCFP-Tree中父節(jié)點(diǎn)的指針。沿指針一直向上搜索，最后指向根節(jié)點(diǎn)。NodeLink：指向FCFP-Tree中同一NodeItem的下一個(gè)節(jié)點(diǎn)的指針。也就是說(shuō)，所有具有相同名稱(chēng)的項(xiàng)目都與FCFP樹(shù)中的NodeLink指針相連。

頭表中的每個(gè)項(xiàng)目由三個(gè)字段組成：NodeItem，NodeCount和NodeLink.頭表中的項(xiàng)目按其支持計(jì)數(shù)按降序排列。

定義2節(jié)點(diǎn)壓縮：如果樹(shù)包含單個(gè)路徑P，則P中的所有項(xiàng)目將被壓縮并存儲(chǔ)在一個(gè)節(jié)點(diǎn)上，該節(jié)點(diǎn)的支持表示為P中任何節(jié)點(diǎn)的支持。

FCFP-Tree構(gòu)造算法類(lèi)似于FP-Tree算法，除了FCFP樹(shù)存儲(chǔ)完整的項(xiàng)集。為了進(jìn)一步壓縮不頻繁項(xiàng)目的存儲(chǔ)，通過(guò)定義2，在單個(gè)分支中具有相同計(jì)數(shù)的不頻繁項(xiàng)目的項(xiàng)目存儲(chǔ)在同一節(jié)點(diǎn)上。

2.2 有效的FCFP-Tree更新策略

在增量挖掘過(guò)程中，項(xiàng)目頻率將不斷改變，即從最初的不頻繁到頻繁或從最初的頻繁到不頻繁。在此過(guò)程中，涉及復(fù)雜的樹(shù)結(jié)構(gòu)更新。

更新FCFP-Tree涉及兩個(gè)步驟：(1)調(diào)整與原始數(shù)據(jù)對(duì)應(yīng)的樹(shù)結(jié)構(gòu)。FCFPIM首先掃描新添加的數(shù)據(jù)集并相應(yīng)地更新頭表，然后根據(jù)頭表中每個(gè)項(xiàng)的相對(duì)位置變化調(diào)整節(jié)點(diǎn)路徑得到調(diào)整后的樹(shù)結(jié)構(gòu)；(2)再次掃描新增的數(shù)據(jù)集以將事務(wù)添加到調(diào)整后的樹(shù)結(jié)構(gòu)，并修改每個(gè)節(jié)點(diǎn)對(duì)應(yīng)的計(jì)數(shù)，得到最終的樹(shù)結(jié)構(gòu)。其中步驟(2)相對(duì)簡(jiǎn)單，針對(duì)步驟(1)，樹(shù)結(jié)構(gòu)調(diào)整包含兩種情況：1)頻繁項(xiàng)目的調(diào)整策略；2)涉及不頻繁項(xiàng)目的調(diào)整策略。每種情況又涉及相鄰節(jié)點(diǎn)和和非相鄰結(jié)點(diǎn)間的調(diào)整，具體調(diào)整見(jiàn)文獻(xiàn)[4].

該兩階段的調(diào)整策略避免了每掃描一條事務(wù)就調(diào)整一次樹(shù)結(jié)構(gòu)的復(fù)雜調(diào)整代價(jià)，而是根據(jù)更新后的頭表一次性調(diào)整好樹(shù)結(jié)構(gòu)后，后續(xù)直接將新增數(shù)據(jù)集涉及到的項(xiàng)目計(jì)數(shù)增添進(jìn)去，從而大大節(jié)省了樹(shù)結(jié)構(gòu)調(diào)整代價(jià)。

3 基于Spark的FCFPIM并行算法及RDD轉(zhuǎn)換策略

3.1 FCFPIM并行算法概述

Spark是一種基于內(nèi)存的迭代式計(jì)算框架，其盡量將中間數(shù)據(jù)放置在內(nèi)存，并將這些數(shù)據(jù)集抽象為RDD(彈性分布式數(shù)據(jù)集)對(duì)象，然后采用一系列的算子來(lái)處理這些RDD，并將處理好的結(jié)果以RDD的形式輸出到內(nèi)存或以數(shù)據(jù)流的方式持久化寫(xiě)入到其它存儲(chǔ)介質(zhì)中。Spark通過(guò)RDD間轉(zhuǎn)換形成的DAG關(guān)系，即Lineage(血緣關(guān)系)來(lái)實(shí)現(xiàn)高效的容錯(cuò)。因此，Spark能很好地適應(yīng)頻繁模式挖掘這種反復(fù)迭代的數(shù)據(jù)處理。

基于Spark的FCFPIM算法由兩個(gè)部分組成，第一部分是對(duì)原始數(shù)據(jù)的挖掘，第二部分是對(duì)增量數(shù)據(jù)進(jìn)行挖掘。基于Spark模型并行化的FCFPIM算法具體過(guò)程描述如下：

Step 1：采用TextFile從HDFS中讀取數(shù)據(jù)集并轉(zhuǎn)化為RDD，將存儲(chǔ)事務(wù)集的RDD記作TransactionRDD.

Step 2：并行計(jì)算原始數(shù)據(jù)1項(xiàng)集的支持度，然后按照支持度降序?qū)?項(xiàng)集排序，得到原始數(shù)據(jù)集頭表，記為DescendItem1.

Step 3：原始數(shù)據(jù)采用關(guān)聯(lián)分組策略分組。將TransactionRDD中存儲(chǔ)的事務(wù)集根據(jù)DescendItem1中的順序進(jìn)行排序，然后按照關(guān)聯(lián)分組策略對(duì)事務(wù)集進(jìn)行分組，得到存儲(chǔ)組號(hào)以及劃分到對(duì)應(yīng)分組的項(xiàng)，記為Group_list.

Step 4：對(duì)TransactionRDD中所存儲(chǔ)的事務(wù)，將每個(gè)項(xiàng)所屬的事務(wù)根據(jù)Group_list的分組規(guī)則劃分到該項(xiàng)所屬的組中，然后對(duì)每一個(gè)組建FCFP樹(shù)，進(jìn)行頻繁模式挖掘。

Step 5：集群各個(gè)節(jié)點(diǎn)的任務(wù)執(zhí)行完畢后，將各部分結(jié)果聚合，得到原始數(shù)據(jù)挖掘結(jié)果，并將結(jié)果輸出到HDFS中，即原始數(shù)據(jù)的挖掘完畢。

Step 6：增量數(shù)據(jù)進(jìn)行挖掘開(kāi)始，讀入新增數(shù)據(jù)，并行計(jì)算該數(shù)據(jù)中的1項(xiàng)集以及支持度，記為NewItem1.

Step 7：將原始數(shù)據(jù)集頭表DescendItem1和NewItem1進(jìn)行對(duì)比判斷，得到需要調(diào)整的節(jié)點(diǎn)以及節(jié)點(diǎn)路徑，通過(guò)廣播的形式分發(fā)到各任務(wù)中。

Step 8：將增量數(shù)據(jù)按照增量部分的關(guān)聯(lián)分組策略，進(jìn)行分組并且將數(shù)據(jù)劃分到對(duì)應(yīng)的分組中，進(jìn)行樹(shù)結(jié)構(gòu)的調(diào)整。

Step 9：對(duì)每個(gè)更新后的FCFP樹(shù)進(jìn)行頻繁項(xiàng)集挖掘，待集群各個(gè)節(jié)點(diǎn)任務(wù)執(zhí)行完畢后，對(duì)結(jié)果進(jìn)行聚合，得到增量數(shù)據(jù)挖掘的結(jié)果，并將結(jié)果輸出到HDFS中。

3.2 有效的RDD轉(zhuǎn)換策略

RDD是Spark的核心，是Spark中最基本的數(shù)據(jù)抽象。RDD 的操作分為轉(zhuǎn)換(Transformation)和動(dòng)作(Action)操作。轉(zhuǎn)換就是從一個(gè) RDD 產(chǎn)生一個(gè)新的 RDD，而動(dòng)作就是進(jìn)行實(shí)際的計(jì)算。 Spark 里的計(jì)算都是通過(guò)操作 RDD 完成的， RDD之間會(huì)形成類(lèi)似于流水線一樣的前后依賴(lài)關(guān)系，即形成邏輯上的DAG，任務(wù)執(zhí)行時(shí)，可以按照DAG的描述，執(zhí)行真正的計(jì)算，因此，合理有效的DAG策略對(duì)算法性能會(huì)產(chǎn)生很大程度的影響。

通過(guò)對(duì)RDD緩存策略的優(yōu)化，對(duì)FCFPIM算法的RDD轉(zhuǎn)換策略進(jìn)行探索，給出一種有效的RDD轉(zhuǎn)換策略。對(duì)應(yīng)于3.1節(jié)描述的步驟，優(yōu)化的RDD轉(zhuǎn)換策略被分為原始數(shù)據(jù)挖掘?qū)?yīng)的RDD轉(zhuǎn)換(見(jiàn)圖1)和增量頻繁項(xiàng)集挖掘的RDD轉(zhuǎn)換(見(jiàn)圖2).

在圖1stage0中，數(shù)據(jù)被劃分為p個(gè)分區(qū)， 并通過(guò)TextFile將數(shù)據(jù)轉(zhuǎn)化為T(mén)ransactionRDD， 接著將數(shù)據(jù)集拆分成各個(gè)事務(wù)轉(zhuǎn)換成SplitTransactionRDD，并且將其緩存到內(nèi)存中。Stage1中的itemRDD通過(guò)reduceByKey算子的操作轉(zhuǎn)化為item1RDD，獲得了1項(xiàng)集且按頻率從高到低對(duì)項(xiàng)目進(jìn)行排序，生成項(xiàng)目和索引的映射DescendItem1Map.Stage2中得到關(guān)聯(lián)分組的結(jié)果AssociationGroup.stage3根據(jù)關(guān)聯(lián)分組的結(jié)果將項(xiàng)目劃分為p個(gè)組，每個(gè)組具有唯一的組ID.對(duì)數(shù)據(jù)SplitTransactionRDD中的每一條事務(wù)按照關(guān)聯(lián)分組的策略生成每組的事務(wù)集，每個(gè)事務(wù)集中的事務(wù)都是完整的，如GenCondTransRDD中。stage4條件模式庫(kù)通過(guò)GroupID縮減為不同的數(shù)據(jù)分區(qū)，每個(gè)分區(qū)都由具有相同GroupID的庫(kù)組成。首先為每個(gè)組生成一個(gè)FCFPTree，并將其存儲(chǔ)在FCFPTreeRDD中。然后，挖掘每個(gè)FCFPTree生成頻繁項(xiàng)FrequentItemRDD.

圖1 原始頻繁項(xiàng)集挖掘RDD轉(zhuǎn)換圖Fig.1 RDD conversion diagram of original frequent itemsets mining

圖2中stage0到stage4對(duì)應(yīng)原始頻繁模式的挖掘。Stage5讀入新數(shù)據(jù)，將事務(wù)進(jìn)行拆分生成SplitIncreTransRDD，緩存到內(nèi)存中，為方便更新樹(shù)的時(shí)候插入新數(shù)據(jù)。并且根據(jù)后面第4節(jié)提出的分組策略將數(shù)據(jù)劃分到對(duì)應(yīng)分組。 Stage6將原始數(shù)據(jù)集的1項(xiàng)集與新增1項(xiàng)集進(jìn)行整合，生成新的1項(xiàng)集NewItem1RDD.Stage7將新的1項(xiàng)集進(jìn)行降序排序，將降序后的新1項(xiàng)集與stage2中的SortItem1Map進(jìn)行對(duì)比，得到需要調(diào)整的節(jié)點(diǎn)以及其調(diào)整的路徑，存儲(chǔ)在AdjustItem中。接著通過(guò)廣播變量的形式，將AdjustItem分發(fā)到對(duì)應(yīng)的節(jié)點(diǎn)上，對(duì)FCFP-Tree進(jìn)行更新。Stage8生成的每個(gè)分組對(duì)應(yīng)的事務(wù)集存儲(chǔ)在GenCondTransRDD中。stage9挖掘每個(gè)FCFPTree生成頻繁項(xiàng)FrequentItemRDD.

圖2 增量頻繁項(xiàng)集挖掘的RDD轉(zhuǎn)化圖Fig.2 RDD conversion diagram of incremental frequent itemsets mining

由圖1和圖2可以看出在實(shí)現(xiàn)FCFPIM算法過(guò)程中，綜合選取避免shuffle和復(fù)雜度較小的算子進(jìn)行RDD的轉(zhuǎn)化以及進(jìn)行了合理的RDD緩存，盡量將多次使用的RDD緩存下來(lái)(如：SplitTransactionRDD)，避免RDD的重復(fù)計(jì)算，在一定程度上提高了RDD緩存命中率。

4 關(guān)聯(lián)分組

在圖1中stage2的AssociationGroup操作和圖2stage5的SplitIncreTransRDD操作都涉及在并行計(jì)算過(guò)程中被計(jì)算項(xiàng)目的分組問(wèn)題。在集群系統(tǒng)中，每個(gè)節(jié)點(diǎn)都可以處理一部分負(fù)載，針對(duì)關(guān)聯(lián)規(guī)則并行算法，項(xiàng)目分組策略一方面會(huì)影響集群中各節(jié)點(diǎn)的計(jì)算負(fù)載，另一方面會(huì)影響數(shù)據(jù)的冗余發(fā)送和存儲(chǔ)。而現(xiàn)有項(xiàng)目分組一般都采用哈希分組，無(wú)法保證集群的負(fù)載均衡，也沒(méi)有考慮數(shù)據(jù)的冗余問(wèn)題。因此，我們提出了關(guān)聯(lián)分組的策略。

FCFPIM算法挖掘頻繁模式仍然采用傳統(tǒng)FP-Growth的遞歸挖掘過(guò)程，考慮到遞歸挖掘過(guò)程中，樹(shù)越高，遞歸所用的時(shí)間越多，而項(xiàng)目支持度越高，對(duì)應(yīng)的事務(wù)數(shù)越多，樹(shù)的高度越低；反之亦然。因此，關(guān)聯(lián)分組策略的基本思想是：為保持集群的負(fù)載均衡，支持度大的項(xiàng)目盡量與支持度小的項(xiàng)目混合分組，同時(shí)為了減少數(shù)據(jù)的冗余存儲(chǔ)，相關(guān)性高的項(xiàng)目應(yīng)該分為一組。由此，頻繁項(xiàng)與非頻繁項(xiàng)得到了相對(duì)均勻的分配。

該關(guān)聯(lián)分組采用了一種壓縮二項(xiàng)集存儲(chǔ)矩陣ArrayItem2，表1為一個(gè)二項(xiàng)集存儲(chǔ)矩陣實(shí)例。

以表1為例，對(duì)關(guān)聯(lián)分組過(guò)程描述如下：

表1 二項(xiàng)集存儲(chǔ)矩陣Tab.1 2-itemset count matrix

Step 1：對(duì)Item1順序分組，每組元素的個(gè)數(shù)為groups個(gè)，記為fristStepGroup ={{a，b}，{c，d}，{e，f}}.

Step 2：取fristStepGroup中的元素fristStepGroup(0) = {a，b}與fristStepGroup(1) ={c，d}，進(jìn)行關(guān)聯(lián)度分組：對(duì)fristStepGroup(0)與fristStepGroup(1)做笛卡爾積運(yùn)算得item2={(a，c)，(a，d)，(b，c)，(b，d)}.

Step 3：在Item2RDD中查找每個(gè)item2的支持度計(jì)數(shù)：(a，c)=5，(a，d)=4，(b，c)=3，(b，d)=3 若最大值只有一個(gè)，那么取最大值的組合分為一組：(a，c).若最大值有多個(gè)，則選取第一個(gè)出現(xiàn)的最大值的組合。

Step 4：去掉item2中帶a和c的組合，重復(fù)Step4步，直到item2的長(zhǎng)度為0.得到的結(jié)果為groups：(a，c)，(b，d).

Step 5：刪除fristStepGroup(0)，若fristStep-Group的長(zhǎng)度不為0，則執(zhí)行第三步得item2={(c，e)，(c，f)，(d，e)，(d，f)}.執(zhí)行Step3 (c，e)=2，(c，f)=1，(d，e)=2，(d，f)=2，得到(c，e)將e加入到(a，c，e).執(zhí)行Step4得到(d，f)將f加入到(b，d，f).

最終，groups={group0={a，c，e}， group1= {b，d，f}}， group0對(duì)應(yīng)數(shù)據(jù)分區(qū)對(duì)項(xiàng)a，c，e是完備的，group1對(duì)應(yīng)數(shù)據(jù)分區(qū)對(duì)項(xiàng)b，d，f是完備的。對(duì)每個(gè)分區(qū)中的事務(wù)分別建立FCFP樹(shù)，并且進(jìn)行頻繁項(xiàng)集挖掘。增量數(shù)據(jù)的新增項(xiàng)目分組一是基于調(diào)整后的二項(xiàng)集存儲(chǔ)矩陣，在此不再贅述，具體分組算法見(jiàn)算法1：

算法1：關(guān)聯(lián)分組

輸入：Item1，ArrayItem2

輸出：groups

0： val item1 = item1RDD.toList

1： val group_num = (item1.count+1 + group_size-1) / group_size

2： //將item1按順序分為group_num個(gè)組，每個(gè)組小于或等于group_size個(gè)元素

3： val fristStepGroup = new ListBuffer[ListBuffer[String]]

4： for(i <-1 to group_num){

5： val k = (i-1)*group_size +1

6： val group_list = new ListBuffer[String]()

7： for (j <- k to Item1.length){

8： if( j % group_num !=0){

9： group_list += Item1(j-1)

10： }else{

11： group_list += Item1(j-1)

12： break

13： End If

14： End For

15： fristStepGroup += group_list

16： End For

17： for (i <- fristStepGroup.length-1){

18： If (i+1 <=fristStepGroup.length-1 ){

19:val item2 = fristStepGroup(i) Cartesian(fristStepGroup(i+1))

20： 在ArrayItem2中查找每個(gè)item2的支持度計(jì)數(shù)

21：得到item2中的最大值(Max._1，Max._2)

22： Gruops+= (Max._1，Max._2)

23：刪除item2中帶Max._1和Max._2的組合

24： End If

25： End for

5 實(shí)驗(yàn)結(jié)果及分析

本次實(shí)驗(yàn)搭建了Spark On YARN的完全分布式集群，在VMware Workstation 上創(chuàng)建三臺(tái)虛擬機(jī)作為集群的節(jié)點(diǎn)，具體配置見(jiàn)表2.

表2 集群系統(tǒng)環(huán)境配置Tab.2 Cluster environment configuration

為了評(píng)估FCFPIM算法的性能，與經(jīng)典的基于Spark的并行FPGrowth算法與最新的基于Spark的并行EFUFP算法進(jìn)行了對(duì)比。數(shù)據(jù)集采用IBM Quest Market-Basket生成器產(chǎn)生，其可以靈活配置以創(chuàng)建各種數(shù)據(jù)集，來(lái)滿足各種測(cè)試需要的需求。

5.1 不同規(guī)模數(shù)據(jù)集對(duì)算法性能的影響

該組實(shí)驗(yàn)采用T10I4D系列和T40I10D系列各4組不同事務(wù)數(shù)的數(shù)據(jù)集(事務(wù)數(shù)分別為10萬(wàn)，20萬(wàn)，30萬(wàn)和40萬(wàn)條)，對(duì)三種算法的運(yùn)行效率進(jìn)行評(píng)價(jià)，其中，增量數(shù)據(jù)的大小大約是原始數(shù)據(jù)大小的10%，最小支持度設(shè)為0.9%.具體實(shí)驗(yàn)結(jié)果見(jiàn)圖3和圖4.

圖3 在數(shù)據(jù)集T10I4D上的執(zhí)行時(shí)間對(duì)比Fig.3 Execution time on the data set T10I4D

圖4 在數(shù)據(jù)集T40I10D上的執(zhí)行時(shí)間對(duì)比Fig.4 Execution time on the data set T40I10D

從圖3和圖4可以看出，F(xiàn)CFPIM算法性能表現(xiàn)最優(yōu)，并行FP-Growth最差，這主要是因?yàn)镕P-Growth在增量過(guò)程中，采用最直接的方式，重新建樹(shù)挖掘，因此消耗的時(shí)間比較長(zhǎng)，而EFUFP算法則在增量挖掘階段采用了高效的調(diào)整策略。本文的FCFPIM算法一方面通過(guò)考慮項(xiàng)目之間的關(guān)聯(lián)程度，對(duì)項(xiàng)目進(jìn)行均衡分組，使得分組后各數(shù)據(jù)分區(qū)對(duì)應(yīng)的冗余存儲(chǔ)大大減少，對(duì)應(yīng)每個(gè)分區(qū)的計(jì)算量亦會(huì)下降，且均衡的分組進(jìn)一步提高了資源利用率，同時(shí)，有效的緩存策略使得一些被多次使用的RDD不再需要重復(fù)計(jì)算，從而進(jìn)一步加速了FCFPIM算法的執(zhí)行。

5.2 不同支持度對(duì)算法性能的影響

本組實(shí)驗(yàn)采用事務(wù)數(shù)為10萬(wàn)條的T40I10D的數(shù)據(jù)集，并將支持閾值分別設(shè)置為0.5%，0.7%，0.9%，1.1%，1.3%進(jìn)行測(cè)試。

從對(duì)比結(jié)果圖5可以看出,隨著支持閾值的增加,生成的頻繁項(xiàng)集數(shù)目會(huì)減少,挖掘的時(shí)間會(huì)減少,這符合理論上的運(yùn)行趨勢(shì)。

圖5 不同支持度的執(zhí)行時(shí)間Fig.5 Execution time with different support

5.3 可擴(kuò)展性

集群的節(jié)點(diǎn)數(shù)量對(duì)算法的執(zhí)行時(shí)間有著最重要的影響，本組實(shí)驗(yàn)采用與上組實(shí)驗(yàn)同樣的數(shù)據(jù)集，支持度計(jì)數(shù)設(shè)置為0.9%進(jìn)行測(cè)試，測(cè)試結(jié)果如圖6所示。

圖6 可擴(kuò)展性Fig.6 Scalability

通過(guò)圖6可以明顯看出，隨著節(jié)點(diǎn)數(shù)的增加，三個(gè)算法整體挖掘時(shí)間均減少，因?yàn)?，可用的?jì)資源隨之增加。在只有一個(gè)計(jì)算節(jié)點(diǎn)的時(shí)候，F(xiàn)CFPIM算法與EFUFP算法的執(zhí)行時(shí)間差不是很明顯，因?yàn)橹挥幸粋€(gè)節(jié)點(diǎn)，分組策略是無(wú)效的，所以?xún)蓚€(gè)算法的執(zhí)行時(shí)間差距不大，完全取決于算法本身的性能與FCFPIM算法的緩存策略。隨著計(jì)算節(jié)點(diǎn)數(shù)目的增加，基于關(guān)聯(lián)分組策略?xún)?yōu)勢(shì)逐漸體現(xiàn)出來(lái)。

5.4 加速比

加速比為并行算法在一個(gè)計(jì)算節(jié)點(diǎn)上的執(zhí)行時(shí)間與在多個(gè)計(jì)算機(jī)節(jié)點(diǎn)上的執(zhí)行時(shí)間的比值。從圖7中可以看出，隨著集群計(jì)算節(jié)點(diǎn)數(shù)量的增加，三個(gè)算法的加速比大致呈現(xiàn)出線性關(guān)系，說(shuō)明三個(gè)并行的算法都是有效的。整體上隨著集群計(jì)算節(jié)點(diǎn)數(shù)目的增加，三個(gè)算法的執(zhí)行時(shí)間差越來(lái)越大。因?yàn)镕CFPIM擁有合理的分組策略。隨著計(jì)算節(jié)點(diǎn)的增加，關(guān)聯(lián)分組的策略會(huì)有明顯的優(yōu)勢(shì)，因此FCFPIM并行算法的性能要更好一些。

圖7 加速比Fig.7 Speedup

6 總結(jié)與展望

針對(duì)傳統(tǒng)增量挖掘存在重復(fù)掃描原始數(shù)據(jù)集或樹(shù)結(jié)構(gòu)調(diào)整成本過(guò)高等問(wèn)題，以及面對(duì)海量數(shù)據(jù)處理傳統(tǒng)串行計(jì)算不足以應(yīng)對(duì)的問(wèn)題，提出一種基于Spark并行框架的并行增量頻繁模式挖掘算法FCFPIM，F(xiàn)CFPIM綜合考慮了負(fù)載均衡，RDD的shuffle成本，緩存策略等問(wèn)題。而由于FCFP樹(shù)包含完整的項(xiàng)目信息，因此該空間開(kāi)銷(xiāo)會(huì)對(duì)算法的可伸縮性產(chǎn)生不利影響。作為未來(lái)的研究方向，我們將引入一個(gè)度量來(lái)確定如何合理的將項(xiàng)目保存在樹(shù)結(jié)構(gòu)中。該度量的定義將考慮數(shù)據(jù)的特征以及新數(shù)據(jù)與原始數(shù)據(jù)的比率。此外，F(xiàn)IM的實(shí)際應(yīng)用也值得關(guān)注。