分布式多數據流頻繁伴隨模式挖掘*

于自強,禹曉輝,董吉文,王 琳

1(濟南大學 信息科學與工程學院,山東 濟南 250022)

2(山東大學 計算機科學與技術學院,山東 濟南 250101)

1 引 言

隨著移動終端和無線網絡的廣泛應用,現實中許多應用面臨大規模數據流的處理.例如,社交網絡中的微博數據,電子商務領域的顧客瀏覽、交易數據,城市交通管控系統的過車數據,環境監測領域的傳感器數據等都包含了大規模數據流.在海量數據流中,通常隱藏著大量用戶感興趣的特定模式,發現這些特定模式對于改善應用服務質量,提升數據應用價值具有重要意義.多數據流中的頻繁伴隨模式是由Yu等人[1]首次提出來的,現實中若干應用都可以抽象為頻繁伴隨模式的發現問題.本文的研究目標是設計分布式挖掘算法,實時發現海量數據流中的頻繁伴隨模式.為便于描述,下文將采用FCP來表達頻繁伴隨模式.

給定一個無界數據流的集合S={s1,s2,…,sn}和一個元素集合O={o1,o2,…,ok},如果集合O是一個 FCP,那么它將滿足以下條件:(1) 該集合中的元素在小于τ的時間段內出現在至少θ個數據流上；(2) 該集合的元素在每一個數據流上出現的時間間隔不大于ξ.這里,τ、θ、ξ都是用戶指定的參數[1].頻繁伴隨模式表達了一組對象在多個數據流中的緊密關系,許多應用都可以歸結為數據流上的頻繁伴隨模式發現問題[1].

伴隨車輛發現.當前城市主要道路和路口安裝了監控抓拍設備,車輛每次經過時,抓拍設備都會生成一條結構化的過車記錄(vehicle passing record,簡稱VPR).這樣,每個抓拍設備將產生一個由連續過車記錄組成的數據流,而一些伴隨行駛的車輛將會出現在由多個抓拍設備所產生的數據流中.因此,從過車記錄中發現伴隨車輛相當于從這些數據流中發現FCP.現實中,及時發現伴隨車輛對于預防和打擊罪犯嫌疑人駕駛多部車輛協同作案、尾隨受害人作案等案件具有重要意義.

基于微博的熱點話題發現.在新浪微博等社交平臺,可以認為每個用戶對應著一個由自己發布的微博組成的數據流.較短時間內,在許多用戶微博中頻繁出現的詞匯組合通常預示著一個新的熱點話題的出現.那么從海量用戶對應的數據流中挖掘頻繁伴隨出現的熱點詞匯組合其本質也是多數據流的FCP發現問題.

位置信息服務.在一些簽到應用中(如 Foursquare),用戶會在到達一個地點之后簽到,并向數據中心報告自己的位置.在這類應用中,可以認為每個簽到的地點對應著一個數據流,而每個數據流是由用戶的簽到信息組成.在這些數據流中實時挖掘 FCP,能夠發現短時間內共同到達某些簽到地點的用戶群,而發現這些用戶具有潛在的商業價值,例如可以作為某些廣告更好地推送對象.

以上事例所要解決的基本問題都是發現較短時間內出現在多個數據流上的FCP(車輛、人、關鍵詞等).與多數據流頻繁伴隨模式發現較為相似的問題是數據流中的頻繁模式發現問題,但兩者有明顯區別.數據流中的頻繁模式是指某一個元素集合在一個數據流的多個“事務(transaction)”中出現,且出現的頻率大于給定的閾值；本文所研究的頻繁伴隨模式則是指一個元素集合在多個數據流中出現,并且所在數據流的數目和出現的間隔需要滿足指定的條件.由于問題本身不同,因此,已有數據流頻繁模式挖掘算法無法直接用來解決本文所研究的問題,原因將在下一節進行詳細闡述.

目前,對于多數據流FCP發現問題的研究還很少.文獻[1]提出該問題的同時,給出了CooMine挖掘算法.該算法的目的是通過減少內存消耗和索引維護代價快速發現多數據流中的 FCP.值得注意的是,CooMine算法是針對單個計算節點設計的集中式挖掘算法,而單個計算節點受到硬件資源的限制,很難應對大規模數據流.為解決這一問題,最直接的辦法是采用多個計算節點,令每個計算節點獨立運行 CooMine算法,處理一部分數據流.該策略雖然能夠處理大規模數據流,但缺點是形成某些 FCP的數據流可能被分在不同的計算節點,導致部分結果丟失.另一策略是將CooMine算法在多個計算節點上并行化實現,但是CooMine算法采用的Seg-tree樹型索引結構很難有效地并行化部署到多個計算節點上.因此,當在分布式計算環境下解決大規模數據流 FCP發現問題時無法直接采用CooMine算法,仍需設計高效的分布式挖掘算法.

與集中式算法相比,設計分布式挖掘算法面臨以下挑戰.

(1) 由于每個計算節點只存儲部分數據,這給 FCP的生成和比對帶來很大困擾.例如,對于一個分布在計算節點n1上的長度為k的FCP,計算節點n1無法得知該FCP應和哪些計算節點的長度為k的FCP合并,生成長度為(k+1)的 FCP.

(2) 采用多個計算節點處理海量數據流時,可能出現不同計算節點上數據負載不均的情況,從而影響整體的處理效率.這是因為,每個 FCP的生成需要多個計算單元協同合作,負載失衡時,負載較輕的計算單元往往需要等待負載較重的計算單元的計算結果,從而影響了整個系統的挖掘效率.

(3) 由于大規模數據流連續到達,新到達的數據不斷地和已有數據共同構成新的FCP,這就要求挖掘算法必須能夠隨著數據流的連續到達實時發現新生成的FCP,實現FCP的增量挖掘.

為應對以上挑戰,本文提出FCP分布式挖掘方法(FCP distributed mining approach,簡稱FCP-DM).該方法的核心思想是采用基于服務器集群的分布式計算模式,從大規模數據流中實時連續發現 FCP.FCP-DM 首先將每個連續到達的數據流劃分成若干 segment片段,將問題轉化為從不同數據流的 segment中發現 FCP；然后基于Actor-Model計算模型構建多級分布式挖掘框架,由多計算節點實現對不同數據流的 segment片段的逐級并行處理；最后根據Apriori算法的迭代思想,設計不同層級的數據分發策略,通過對segment的多層計算和比對,最終得到FCP.

本文的主要貢獻如下:

· 提出 FCP-DM 分布式挖掘方法并給出相應的分布式多級計算框架,能夠實現對大規模數據流的并行處理,通過多層迭代計算和比對發現所有的 FCP.FCP-DM 方法具備良好的可擴展性,僅通過增加硬件資源就可實現處理能力的線性增長.

· 解決了分布式環境下多數據流FCP挖掘所面臨的負載遷移、多計算節點之間FCP分發策略以及面向連續數據流的FCP增量挖掘等問題.

· 在開源流數據處理平臺S4上實現了FCP-DM方法,并且以山東省會城市濟南某天所有卡口的過車記錄為測試數據集進行大量實驗,充分驗證了該算法的各項性能.

2 相關工作

多數據流頻繁伴隨模式發現問題是由Yu等人首次提出來的[1],并給出兩種集中式挖掘算法:DIMine算法和CooMine算法.這兩種算法通過設計不同的索引結構來提高挖掘效率并達到節省存儲空間的目的.然而,這兩種算法都是基于單機計算環境而設計,難以直接應用到分布式環境中,可擴展性較差,無法應對當前規模急劇增長的海量數據流.

已有工作中,與多數據流 FCP發現問題最為相似的是數據流上的頻繁模式挖掘問題.數據流上的頻繁模式(frequent pattern,簡稱FP)[2]是指給定一個由連續的transaction組成的數據流(每個 transaction包含多個元素),如果指定的某段時間內該數據流共有n個transaction,而一個元素集合在其中的m(m≤n)個transaction中出現并且,那么該元素集合就是一個頻繁模式,sup是用戶指定的閾值.下文采用FP表示頻繁模式.

近年來,流數據FP挖掘問題受到國內外學者的廣泛關注[2-12].Manku和Motwani提出了粘性抽樣和有損耗的計數算法來計算數據流元素集合的近似頻率[4].Yu等人提出了 FDPM算法[5],該算法能夠使用有界的內存挖掘數據流中的頻繁元素集合.該算法是以假的消極結果為導向的,即某些特定的頻繁元素集合可能不會被發現.這兩項工作都是基于landmark模型,它們的目標都是挖掘數據流從開始到當前時間所有的頻繁模式.此外,它們并不保證發現所有頻繁模式,而是保證獲得一個錯誤率小于指定參數的近似結果集.

另外一些方法[2,5-12]是能夠從數據流中獲得當前精確的頻繁模式集合.Chang等人[2]提出了一種挖掘算法,該算法通過自適應地減小過期事務的影響從而從在線數據流中發現頻繁模式.Leung和 Khan等人[6]提出樹型索引結構(DSTtree),該索引能夠從數據流中捕獲重要信息從而精確地挖掘頻繁模式.Mozafari等人則將數據流劃分成滑動窗口,提出 versification這一新的計數概念,并基于此概念設計了 SWIM 算法.該算法是一種精確算法,其性能和擴展性能夠根據窗口的大小進行調整[8].KARP等人提出了一個簡單、精確的數據流上的頻繁項集發現算法,并證明了該算法的時間復雜度為線性且空間復雜度為 1/θ,θ為用戶指定的支持度[9].Chi等人[10]主要研究內存受限情況下的數據流上閉合頻繁項集挖掘問題,并給出 Moment(maintaining closed frequent itemsets by incremental updates)算法對數據流的閉合頻繁項集進行持續監測.Silva等人提出了Star FP Stream算法,用以解決從多維數據模型所產生的海量星型數據模式中發現頻繁模式[11].李海峰等人研究了數據流上頻繁項集挖掘時所采用的滑動窗口模型,提出了基于事務的可變窗口滑動模型,并在此基礎上提出了頻繁項集的挖掘算法FIMoTS[12].

以上這些方法雖然能夠從數據流中挖掘頻繁模式,但是無法被直接用來解決FCP挖掘問題.首先,FP和FCP的定義有著很大的不同.FP挖掘問題是關注一個元素集合在某個數據流上出現的頻率,而FCP挖掘問題則關注一個元素集合在數據流內部和多個數據流之間的伴隨關系.此外,上述方法主要針對單個數據流的 FP挖掘問題,而本文要解決的是多數據流FCP發現問題,需要對多個數據流同時處理,兩者之間存在明顯區別.

值得注意的是,文獻[13]提出了H-stream算法,目標是發現多個數據流上的頻繁模式.表面上看,H-stream算法要解決的問題似乎與本文的問題非常相似,但事實上有很大不同.雖然 H-stream算法的目標是挖掘在多個數據流出現的FP,但本質上還是先查找每個數據流的FP,然后篩選符合條件的FP.而本文中,FCP是由多個數據流共同生成,因此必須同時處理多個數據流才能發現FCP.此外,H-stream是一種近似算法,而本文的目標是求解精確結果.因此,H-stream算法并不能解決本文研究的問題.此外,毛宇星等人雖然也提出一種多層關聯規則挖掘方法[14],但是分層的目的是對每層的數據進行聚類,而本文所設計的多層挖掘模型的目的是充分發揮多計算節點在處理多數據流時的并行計算能力,研究的側重點有很大不同.

頻繁情節挖掘(frequent episode mining)是在頻繁模式挖掘基礎上衍生出來的又一問題,它是指從一些序列數據中發現有價值的或者用戶感興趣的模式序列[15,16].目前,已有若干工作研究多個行業(如通信、制造業、金融、生物信息)的頻繁情節挖掘問題,但是絕大部分工作都是采用離線方式挖掘頻繁情節,這些方法也難以用于實時發現多數據流的頻繁伴隨模式.雖然Ao等人提出了基于episode trie的在線頻繁情節挖掘方法[15],然而該方法所關注的頻繁情節也是基于單個數據流定義的,與本文研究的問題有很大區別.此外,該方法是針對單機計算環境所設計,episode trie索引結構很難直接部署到分布式計算環境,也難以解決本文所研究的問題.

當前,也有一些學者開始研究大數據環境下的模式發現[17-20],但是這類工作通常關注的是靜態數據集上的模式發現問題,而本文關注的是從海量動態數據流上發現特定模式.

3 問題定義

為便于算法描述,本文引入并使用文獻[1]所給出的相關定義.

定義 1(數據流(data stream)).數據流是一組連續的按時間順序到達的元素序列.對于數據流中的任意元素oi(i為該元素在數據流中的序列號),它的標識符和時間戳分別是idi和ti,其中,ti表示該元素出現的時間.

定義2(伴隨模式(co-occurrence pattern,簡稱CP)).給定數據流si上的一個元素集合O={o1,o2,…,ok},如果,則認為集合O是一個伴隨模式CP,這里,,ξ是用戶指定的閾值.CPk(k≥2)表示長度為k的CP,即該CP包含k個元素.

定義3(頻繁伴隨模式(frequent co-occurrence pattern,簡稱FCP)).如果一個伴隨模式CP在l個數據流中出現,l個數據流可以表示為集合S={s1,s2,…,sl}.如果該CP滿足以下條件:

(1)l≥θ；

定義4(segment片段).給定一個數據流si,一個segment片段G(o1,o2,…,om)是數據流si的子序列,并且G滿足以下條件.

(1) |ti-tj|≤ξ,這里,oi,oj是G中任意兩個元素,ti,tj分別是oi,oj出現的時間；

(2) 數據流si中不存在子序列G?,使得G?是一個segment片段并且G是G?的一個嚴格的子序列.

圖1給出由多個元素構成的一個數據流,假設每個元素出現的時間如下:ta=2,tc=5,td=13,tg=16,te=20,tb=26,并假設ξ=10,那么該圖中{a,c,d}和{d,g,e}為兩個 segment.元素a是 segment片段G0的第 1個元素,因為td-ta＜ξ,tg-ta＞ξ,所以元素d是G0的最后一個元素.當以元素c為第 1個元素構建 segment時,由于tg-tc＞ξ,{c,d,g}無法構成segment.進一步地,我們以元素d為第1個元素構建segment時,便可得{c,d,e}是一個segment.

Fig.1 Segment partition圖1 Segment片段

4 FCP-DM挖掘方法

引入 segment之后,每個數據流被劃分成多個 segment片段,那么 FCP-DM 的目標就是從海量數據流的segment中發現FCP.根據CP和FCP的定義,可以推斷出一個數據流的某個segment中的任意CP都可能與其他數據流的segment中的CP形成FCP.因此,為了得到精確結果,FCP-DM算法需要對所有數據流的segment所包含的CP進行比對.

4.1 FCP-DM算法

FCP-DM算法中,每一個數據流首先被劃分成多個segment.由于segment分布在不同的計算節點上,那么如何比對不同segment中的CP是一個首先要解決的問題.

在本文構建的分布式計算環境中,一個物理節點可以運行若干個邏輯計算單元(processing element,簡稱PE).FCP-DM算法的思想是以CP本身為key構建分布式哈希索引(DH-index),每個CP對應DH-index的一個索引單元.DH-index中每個索引單元的value值包括對應CP的出現時間以及它所在的segment和數據流的相關信息.這種情況下,不同segment中相同的CP被插入到DH-index時,這些CP將被映射到同一個索引單元.當FCP-DM被部署到分布式計算環境時,令物理節點的一個PE負責一個索引單元,該索引單元每增加一條新的記錄(即來自某個segment的CP),PE都會判斷該索引單元對應的CP是否為FCP.某個CP一旦滿足FCP定義的條件,FCP-DM會實時發現該FCP.

在圖 2中,給出分別屬于數據流(s1,s3,s2)的 3個 segment(G0,G1,G2),根據給定的 segment,得到所有長度為 2的CP集合.根據圖2的CP2集合建立DH-index索引,如圖3所示.FCP-DM算法對DH-index進行掃描,可以得到全部長度為 2的 FCP(FCP2).例如,設定θ=3,那么 FCP-DM 只需對{o1,o2}進行判斷.如果{o1,o2}出現在s1,s3,s2的時間間隔小于τ,那么{o1,o2}就是一個FCP.FCP-DM算法在獲得FCP2后,便利用Apriori啟發式思想[3],逐步地由FCPk(k≥2)的集合推導出FCPk+1(k≥2)的集合.

Fig.2 Segments and the correspondingCP2 collections圖2 生成segment的CP2集合

Fig.3 The structure of DH-index圖3 DH-index示意圖

實際應用中,海量數據流將產生規模巨大的 CP,超出了單個計算節點的處理能力.為此,本文將 FCP-DM 算法部署到分布式計算環境,通過利用多計算節點的存儲和計算資源,采用并行計算和比對方式,實現海量數據流中FCP的實時發現.下面介紹FCP-DM算法的分布式計算框架.

4.2 FCP-DM分布式計算框架

本文所提出的 FCP-DM 分布式計算框架(FCP-DM distributed framework,簡稱 FCP-DMDF)采用 Actor-Model分布式計算模型,FCP-DMDF中基本邏輯處理單元是 PE(processing element),每個物理節點可以運行任意多個 PE.PE之間通過發送和接收事件(event)進行數據傳輸.一個 Event被表示為〈type,key,value〉,type表示Event的類型,key和 value分別表示 Event的鍵值和內容.每個 PE指定其所接收的 Event的 type和 key值.FCP-DMDF根據PE所指定的type和key值為其分發Event.任意一個PE可以接受其他PE發送的Event,也可以將本地計算結果以Event的形式發送給其他PE.對于一個新的Event,如果已有的PE無法對其進行處理,那么FCP-DMDF將自動構建一個新的PE處理該Event.

FCP-DMDF是一個多級框架,主要包括數據接收層、數據分割層和算法實現層.圖4是FCP-DMDF框架示意圖.

Fig.4 The framework of FCP-DMDF圖4 FCP-DMDF示意圖

數據接收層的任務是接收持續到達的數據流.每個數據流由一個ReceivePE負責,ReceivePE的任務是接收數據流并隨時間延續將每個數據流分割成若干segment,然后將segment以SegmentEvent的形式發送至數據分割層的PartitionPE.

數據分割層的任務是生成每個segment所有長度為2的CP.該層中每個PartitionPE的任務是接收來自不同數據流的segment,并計算每個segment包含的所有長度為2的CP.然后以CP本身為key值將每一個CP以CPEvent的形式分發至算法實現層的PE.在該層中,每個PartitionPE負責一個或多個segment,并能夠根據負載對segment進行數據遷移,第4.3.1節將詳細討論該問題.

算法實現層是 FCP-DMDF的主體部分.該層包含多個Computing-Level,Computing-Level-i表示第i層Computing-Level,負責發現長度為(i+1)的FCP.FCP-DM算法在產生長度為i的FCP時包含兩步操作:CPi的聚合和FCPi的檢測,因此,每層Computing-Level-i包含兩類PE:CPi-PE和FCPi-PE.當i≥3時,由FCPi-1集合得到FCPi集合的迭代計算步驟如下.

①CPi-PE將可能構成CPi的一組FCPi-1進行聚合,生成待檢測的CPi,并將符合條件的CPi以CPEvent的形式發送至FCPi-PE.分布式環境下,如何將可能構成CPi的一組FCPi-1由不同的PE匯聚至同一個CPi-PE是一個不小的挑戰,為此,本文設計了基于〈key,value〉的FCP分發策略(該策略將在第4.3.2節詳細加以討論)；

②FCPi-PE接收來自不同CPi-PE的CPi進行檢測并判斷其是否為FCPi,然后將發現的FCPi以FCPEvent的形式發送至Computing-Level-(i+1)層的CPi+1-PE.

③ 重復步驟①和②,CPi+1-PE將接收的FCPi合并生成CPi+1,然后由FCPi+1-PE檢測CPi+1是否為FCPi+1.根據上述步驟,隨著i的增長,算法實現層能夠通過逐步迭代的方式發現所有的FCP.

需要說明的是,當i=2時,由于CP2由PartitionPE直接生成,因此,每個CP2-PE接收到的是一組來自不同數據流的CP2,CP2-PE將該組CP2進行聚合后發送至FCP2-PE.由于現實應用中 FCP的最大長度無法預先確定,因此FCP-DM計算框架將根據FCP的長度自適應地增加或減少Computing-Level的層數,從而發現給定數據流中所有FCP.

4.3 構建FCP-DMDF面臨的挑戰與解決方案

本節主要研究分布式環境下多數據流FCP挖掘所面臨的負載遷移、FCP分發策略以及面向連續數據流的FCP增量挖掘等挑戰.

4.3.1 數據分割層的負載均衡策略

數據分割層將生成每一個segment片段所有的長度為2的CP,并將CP進行分發.實際應用中,數據分割層的每個計算單元可能負責若干個 segment,如果數據分布不均勻,將造成不同計算單元之間的負載失衡,降低整個計算框架的挖掘效率.

為解決這一問題,本文設計動態遷移策略來保證計算單元之間的負載均衡.為便于描述,令PEr表示數據接收層的任意一個 ReceivePE,令表示數據分割層任意兩個不同的 PartitionPE.此外,令[wh,wk]表示元素序列號的范圍,如果某個segment片段首個元素oi的idi∈[wh,wk],那么該segment位于范圍[wh,wk].

(1) 對PEip的 key值范圍[wh,wk]進行拆分,生成[wh,wj]和[wj,wk](wh

(2)PEpi將[wj,wk]范圍內未處理的segment以SegmentEvent的形式發送至數據分割層的新的邏輯計算單元將由FCP-DMDF在發出SegmentEvent后創建.可接收SegmentEvent的key值范圍為[wj,wk].

圖5是負載遷移執行過程的示意圖.其中,圖5(a)表示初始狀態下數據輸入層的PEr根據的key值范圍向其發送相應的segment.圖5(b)的①②③步則表示上述負載遷移3個步驟的執行過程.其中,

③PEr將后續到達的位于[wj,wk]的segment發送至.

Fig.5 The procedure of workload transfer圖5 負載遷移示意圖

4.3.2 分布式環境下FCP分發策略

FCP-DM算法逐步地由FCPk的集合推導出FCPk+1的集合,這在單機環境下不難實現.但在分布式環境下卻面臨新的問題.

(1) 由于數據分布在多個計算節點上,單個計算節點無法確定哪些FCPk能夠合并生成FCPk+1.

(2) 為解決第1個問題,需要將能夠合并生成FCPk+1的FCPk分發至同一個計算單元,但是如何將這些FCPk分發至同一個計算單元則又是一個新的挑戰.例如,{a,b,c}和{b,c,d}是兩個長度為3的FCP,FCP-DM算法將會對{a,b,c}和{b,c,d}進行合并,生成長度為4的CP{a,b,c,d},繼而判斷{a,b,c,d}是否為FCP4.分布式環境下,{a,b,c}和{b,c,d}可能分布在不同的計算節點n1和n2上,n1和n2無法直接獲取對方的數據,這為生成伴隨模式{a,b,c,d}增加了很大難度.

為解決上述問題,本文設計基于〈key,value〉的 FCP分發策略.該策略首先令能夠合并生成CPk+1的FCPk具有相同的key值,然后令每個key值由FCP-DMDF中唯一的計算單元負責.此時,根據key值與計算單元的映射關系,可將具有相同key值的FCPk分發至同一個計算單元,繼而由該計算單元對這些FCPk進行聚合,生成相應的CPk+1,最后判斷CPk+1是否為FCPk+1.

假設FCPk和FCP?k分別表示兩個不同的長度為k的FCP,FCPk+1表示一個長度為(k+1)的FCP.如果,那么FCPk和FCP?k至少含有(k-1)個相同元素.將FCPk的任意一個由(k-1)個不同元素構成的集合記作ek,那么FCPk將包含個不同的集合.對于任意集合,可能存在一個頻繁伴隨模式滿足,這種情況下,可能產生CPk+1.因此,對于任意一個FCPk,以每個為key,以FCPk自身為value生成k個形如 的元組,并將每個元組根據分發到不同的PE,這使得含有相同的不同元組將被發送至同一個PE.該類PE接收到這些元組之后,能夠將這些元組進行聚合,構成CPk+1,并將該CPk+1發送至下一層計算單元由其判斷該CPk+1是否為FCPk+1.

4.3.3 FCP連續增量挖掘方法

由于數據流持續到達且沒有邊界,新的數據流到達之后,可能與已有數據形成新的 FCP,因此需要對連續到達的數據流持續監控,設計增量挖掘方法實時發現由新到達的數據流所形成的 FCP.在增量挖掘方法中,我們始終維護FCP-DMDF框架,并將已處理的有效數據始終在內存中進行維護.新的數據流到達之后,只需將這些數據與已有的有效數據進行比對,即可發現新生成的FCP.

由于數據流的規模無限擴大,導致無法在內存存儲所有數據,需要對過期數據進行刪除.由于形成FCP的所有元素的時間間隔不大于τ,如果某個元素的產生時間與當前時間的間隔大于τ,那么該元素可以被安全刪除,原因是它不可能和其他元素共同構成新的FCP.從節省內存角度考慮,則應該實時刪除過期元素.然而,如果令每個計算節點持續監視其維護的每條數據是否過期,那么該操作將產生很大的計算代價.為使內存使用效率和計算代價兩方面達到一個平衡,本文設計了延遲刪除策略.該策略中,每個計算節點每隔Δt時間令其所有的邏輯計算單元執行一次對過期數據的刪除操作.實驗部分,我們將Δt的默認值設置為τ.Δt值越大,刪除操作的計算代價越小,但是內存消耗增大；Δt值越小,內存消耗越少,但刪除操作代價增大.因此,可以調整Δt的值來平衡內存使用效率和刪除計算代價.實驗部分將對Δt對于算法內存使用情況的影響進行測試.

由于 FCP-DMDF始終在內存中維護有效數據,那么對于一個最新到達的 segment,如果它能夠和已有數據形成FCP,那么只需要在已有數據的基礎上處理該segment,便可發現由新到達的segment與已有segment所形成的所有FCP,從而實現多數據流FCP的連續增量挖掘.為解決這一問題,本文設計了Incremental-mining算法并給出算法的偽代碼.對于一個新生成的segment,Incremental-mining算法首先生成該segment的所有CP2,之后對應的PE將CP2與已有數據進行比對,得到FCP2.然后,基于Apriori啟發式思想,由對應的PE對FCP2所能構成的 FCP進行逐級并行檢測,從而得到該 segment與已有數據形成的所有 FCP.在 Incremental-mining算法對segment的處理過程中,僅有少量PE參與計算,因此可節省計算資源.

Incremental-mining算法.

圖6給出在FCP-DMDF上利用Incremental-mining算法發現FCP3的一個例子.在該例子中,假設數據流S1至Si的segment已被處理,此時數據流Si+1產生一個新的segment(Gi+1),下面是利用Incremental-mining算法發現由新產生的 segment所形成的 FCP.首先由 ReceivePE將每個數據流劃分成不同的 segment；對于一個segment,PartitionPE生成其包含的所有CP2.CP2-PE將來自不同數據流的相同CP2發送至FCP2-PE,然后由FCP2-PE判斷某個CP2是否為FCP2.FCP2-PE根據第 4.3.2節介紹的分發策略,將得到的FCP2發送至對應的CP3-PE,每個CP3-PE將收到的FCP2進行聚合,生成CP3,最后由FCP3-PE檢測CP3是否為FCP3.

Fig.6 Procedure of miningFCP3 (θ=3)圖6FCP3挖掘示意圖(θ=3)

4.3.4 Incremental-mining算法時間復雜度分析

分布式環境下,Incremental-mining算法處理一個長度為m的 segment片段G的時間復雜度的下界為,最壞情況下的時間復雜度為,z表示運行Incremental-mining算法的一層CPi-PE或FCPi-PE的數量.

證明:Incremental-mining算法處理G時,首先生成個CP2,并檢測每一個CP2是否為FCP2.如果每一個CP2均不是FCP2,則算法終止.為了便于描述,假設個CP2均勻分布到z個PE上進行并行處理,那么每個PE負責CP2的數量約為,單個PE的計算時間為(td為單個CP2的檢測時間).此時的時間復雜度為.

下面討論Incremental-mining算法最壞情況下的時間復雜度.當G能夠形成FCPm時,Incremental-mining算法的計算時間最長.一般情況下,只有部分伴隨模式能夠構成頻繁伴隨模式.不失一般性,假設G的長度為i的子集是FCPi的概率為p(0

雖然Incremental-mining算法最壞情況下的時間復雜度為,但它的實際計算時間要小很多,原因如下.

(1) 真實情況下p值非常小,因此需要檢測的FCPi數量遠小于,即；

(2) Incremental-mining算法運行在多個物理計算節點之上,能夠利用大量的PE實現對FCPi的并行檢測,將大大縮短計算時間.因此,物理節點越多,總的 PE的數量越多(即z值越大),計算時間越短,這與實驗部分中圖10所示結果一致.

5 實 驗

首先,本文基于Apache開源流數據處理平臺S4[21]實現了FCP-DMDF.S4中的邏輯處理單元稱為PE,每個物理節點可以運行任意多個PE.基于S4平臺實現FCP-DMDF時,FCP-DMDF中PE的功能可以由S4的PE來實現.S4中PE之間通過發送和接收Event進行數據傳輸.每個PE指定其所接收的Event的類型和key值.任意一個PE可以接受其他PE發送的Event,也可以將本地計算結果以Event的形式發送給其他PE.對一個新產生的Event,如果已有的PE根據Event的類型和key值都無法處理該Event,那么S4將自動創建一個新的PE來處理該Event.由于S4的并行處理思想被當前大多數流數據分布式計算平臺所采用,因此,本文所設計的FCP-DM分布式挖掘方法也易于部署到Storm、Spark Streaming等平臺.

5.1 實驗配置

實驗采用8臺戴爾 R210的服務器,每臺服務器配置主頻2.4GHz的Intel處理器和8G內存,服務器之間通過1Gbps帶寬的以太網相連.分布式平臺采用S4-0.6.0版本(http://incubator.apache.org/s4/download/).以山東省濟南市2015年5月1日7:00～12:00產生的320萬條過車記錄(vehicle passing records,簡稱VPR)為測試數據集,從中挖掘頻繁伴隨車輛.該數據集中每條過車記錄可以看作是一個四元組〈ri,li,mi,tmi〉,ri表示該記錄的編號,li表示該車的車牌號,mi表示該記錄所對應的監控卡口編號,tmi表示該記錄產生的時間.整個測試集中過車記錄是由不同卡口產生的,如果將同一個卡口連續產生的過車記錄看成一個數據流,那么若干卡口就對應若干數據流.實驗中要查找的伴隨車輛是指很短時間內連續通過某卡口,并在一段時間內以同樣方式通過多個卡口的一組車輛.因此,在測試數據集中發現伴隨車輛就相當于從多個卡口產生的數據流中發現 FCP.FCP定義中相關參數(ξ、τ和θ)將對 FCP-DM 算法的效率和挖掘結果產生影響,因此,每組實驗均標明各個參數的取值.缺省情況下,ξ=60(s),θ=4,τ=2(h).每組實驗均重復5次,取5次結果的平均值作為最終實驗結果.

5.2 實驗內容

5.2.1 FCP-DM算法效率評估

圖7將測試數據集模擬為523個數據流,分別以20 000VPR/s、40 000VPR/s、60 000 VPR/s、80 000 VPR/s、100 000 VPR/s的速率發送至FCP-DMDF,以測試系統的抗壓性.實驗結果表明,FCP-DM算法的最大處理能力約為60 000條/s.在該組實驗中,我們在內存中設置一個隊列Q用以緩存到達的VPR.FCP-DM(t)表示FCP-DM處理測試數據集所需總的時間,FCP-DM(Q)表示隊列Q緩存 VPR的最大數量.在圖 7中,當數據到達速率小于60 000條/s時,隨著數據速率的提高,FCP-DM(t)明顯下降.此時,由于FCP-DM處理速率大于數據到達速率,所以FCP-DM(Q)趨近于 0.當速率達到 60 000 VPR/s后,FCP-DM(t)趨于平穩,表明算法是以最大處理能力運行的,FCP-DM(Q)開始明顯增加.

圖8表明,Yu等人提出的CooMine算法的最大處理能力約為8 000VPR/s(單個計算節點)[1].通過圖7和圖8的比較可以發現,FCP-DM在8個計算節點上的處理能力幾乎是CooMine算法的8倍.也就是說,FCP-DM在8個計算節點的處理效率幾乎相當于在8個計算節點上分別運行CooMine算法的效率總和.這是因為,CooMine算法是針對單機環境設計的多數據流頻繁伴隨模式發現算法,該算法為了節省內存(壓縮 segment)和減少索引維護代價,設計和采用Seg-tree樹型索引結構.本文設計的FCP-DM算法是針對分布式計算環境,擁有較充裕的內存,因此采用了分布式哈希索引.雖然 CooMine算法對基于 Seg-tree的數據查找操作進行了大量優化,但是FCP-DM算法中基于哈希索引的查詢操作(如查找元素、伴隨模式等)要比CooMine算法中基于Seg-tree的查詢操作具備更高的查詢效率,而哈希索引的內存使用效率要低于 Seg-tree索引結構.因此,在 8臺機器上運行FCP-DM算法的計算效率接近于在8個計算節點上單獨運行的CooMine算法的計算效率之和.

Fig.7 Processing capability of FCP-DM圖7 FCP-DM方法處理能力

Fig.8 Processing capability of CooMine圖8 CooMine算法處理能力

為更加直觀地比較FCP-DM算法和CooMine算法的性能,分別令這兩種算法處理相同規模的過車記錄數據集,并調整過車記錄數據集的規模來比較兩者的處理時間.該組實驗是將整個數據集一次性地加載到系統中,算法處理時間是指從處理第 1條數據開始至最后一條數據處理完成的時間跨度.圖 9所示的實驗結果表明,FCP-DM算法(采用8個物理節點)的處理時間要明顯小于CooMine算法的處理時間,并且隨著數據規模的擴大,FCP-DM算法處理時間的增長速率明顯小于CooMine算法的時間增長速率.

圖10測試數據到達速率為40 000條/s時FCP-DM算法處理不同規模數據時的可擴展性.該組實驗中,分別采用2、4、6、8個計算節點處理不同規模的數據集.實驗結果表明,隨著計算節點個數的增加,FCP-DM的處理時間明顯下降.VPR數量越多,處理時間的下降趨勢越明顯,表明FCP-DM方法在處理大規模數據流時具備良好的可擴展性.

Fig.9 Comparison of FCP-DM and CooMine w.r.t. processing time圖9 FCP-DM與CooMine處理時間比較

Fig.10 The scalability of FCP-DM圖10 FCP-DM的可擴展性

5.2.2 FCP-DM算法相關參數對算法性能影響的評價

圖11和圖12主要對本文所采取的負載遷移策略的性能進行驗證,主要測試單個PartitionPE所能承載的segment的最大數量β、數據到達速率以及數據集規模對于 PartitionPE數量的影響.該組實驗共采用4個物理計算節點.實驗開始之前,我們對測試數據集中的所有元素的標識進行遍歷,得到一個能夠包含測試數據集中所有元素的標識范圍[ws,we].實驗開始時,初始化一個PartitionPE,令其key值范圍為[ws,we],也就是說,所有初始的PartitionPE能夠接收所有的 segment.當初始 PartitionPE接收但來不及處理的 segment數量達到閾值β時,該PartitionPE就會分裂,這樣就會產生多個 PartitionPE.這里需要注意的是,PartitionPE一旦創建就不會被刪除,本組實驗將記錄處理過程中所有的PartitionPE的數量.

圖11的實驗結果表明,當β值一定時,隨著數據到達速率的增加,PartitionPE的個數明顯增多,這是因為數據到達越快,PartitionPE所積壓的未處理的 segment越多,這樣發生負載遷移的次數越多,導致更多的 PartitionPE產生.當數據到達速率一定時(例如30 000VPR/s時),隨著β值的增大,PartitionPE的數量逐漸減少.這是因為隨著β值的增大,PartitionPE能夠緩存的segment增多,使得PartitionPE的數量減少.圖12中,令FCP-DM處理不同規模的數據集并觀察PartitionPE的數量.當數據到達速率和β值固定時,我們發現數據集的規模對PartitionPE的數量沒有顯著影響,這表明PartitionPE的數量只與數據到達速率和β值有關,與待處理的數據集規模無關.

Fig.11 Number of PartitionPE w.r.t.β圖11 參數β對PartitionPE個數的影響

Fig.12 Number of PartitionPE w.r.t. scale of data圖12 數據集規模對PartitionPE個數的影響

在延遲刪除策略中,調整參數Δt的值可以改變對過期數據的刪除周期.圖13測試了參數Δt對于算法運行時內存使用狀況的影響.實驗策略是隨機抽取100萬條VPR作為測試數據集,令數據發送速率為10 000VPR/s,然后觀察不同時刻的所有計算節點的內存使用狀態.圖13的縱坐標表示所有計算節點運行FCP-DM算法時消耗的內存之和,橫坐標表示算法的運行時間.實驗結果表明,算法運行初期 Δt取不同值時,內存使用量均增大,這是因為此時內存中沒有過期數據,致使內存累積的數據增加.當算法運行50s之后,內存使用量出現明顯波動,這是因為算法每隔 Δt時間都會對過期數據進行刪除.總的來說,Δt取值越小,平均的內存使用量也越少,但是刪除次數也相對增加.圖14測試了參數ξ對FCP-DM處理時間的影響.圖14所示實驗結果表明,算法處理時間隨著ξ值的增大而增加.這是因為,ξ越大,segment長度越大,此時所產生的CP數量也越大.由于FCP-DM需要對所有的CP進行處理,因此,FCP-DM算法的處理時間隨著CP數量的增大而明顯增加.

FCP-DM算法的目標是準確發現給定數據集中所有的FCP,即查全率和查準率應為100%.由于CooMine算法的目標也是準確發現給定數據流中所有的FCP[1],其查全率和查準率均為100%.表1比較了FCP-DM算法和CooMine算法在同一數據集上所發現的FCP數量.當參數θ和k取值一定時,兩種算法在該實驗數據集上發現的FCP數量相同,從而驗證了FCP-DM算法的查全率和查準率均為100%.

Fig.13 Memory consumption w.r.t. Δt圖13 參數Δt對于內存使用的影響

Fig.14 Processing time w.r.t.ξ圖14ξ對處理時間的影響

Table 1 Comparison of FCP-DM and CooMine w.r.t. the number of discovered FCP表1 FCP-DM和CooMine挖掘結果數量比較

5.2.3 FCP-DM挖掘結果評價

圖15和圖16主要測試數據規模和參數θ對FCP-DM算法挖掘結果的影響,其中,k表示FCP的長度,ξ、τ和θ的含義見定義2和定義3.該組實驗中,一個FCP代表一組頻繁伴隨車輛.圖15所示實驗結果表明,當FCP定義中相關參數(ξ、τ、θ)的值固定時,隨著數據規模的擴大,FCP-DM發現的FCP數量逐漸增多；當數據規模一定時,FCP的長度(k值)越大,FCP的數量越少,也就是說,用戶指定一組頻繁伴隨車輛中的車輛數目越大,現實中這樣的車輛組合越少,越符合我們的直觀認識.FCP定義中參數θ將對FCP的數量產生較大影響,圖16所示實驗結果表明,隨著θ值的增大,FCP數量變小,也就是說,如果用戶要求一組 FCP伴隨出現的數據流越多,那么這樣的FCP數量越少,實驗結果同樣符合預期.目前,FCP-DM算法已被應用至山東省某地市智能交通綜合管控平臺,用于快速發現頻繁伴隨車輛.圖17和圖18給出了FCP-DM算法在該交通管控平臺發現頻繁伴隨車輛的部分結果展示.其中,地圖中的紅色箭頭是車輛伴隨行駛方向,圖片右側是發現的頻繁伴隨車輛組合.

Fig.15 Number of FCP w.r.t. scale of data圖15 數據規模對FCP數量的影響

Fig.16 Number of FCP w.r.t. θ圖16 參數θ對FCP數量的影響

Fig.17 Demonstration of applying FCP-DM (1)圖17 FCP-DM實際應用展示(1)

Fig.18 Demonstration of applying FCP-DM (2)圖18 FCP-DM實際應用展示(2)

6 總 結

本文研究了海量多數據流 FCP挖掘問題,提出了 FCP-DM——一個可部署在分布式流數據處理平臺的分布式挖掘方法.在該方法中,本文重點研究分布式環境下多計算節點協同挖掘大規模數據流所形成的 FCP時面臨的負載均衡問題、數據分布式存儲背景下的FCP生成問題以及連續數據流的FCP增量挖掘問題,最后通過大量實驗對 FCP-DM 方法的各項性能和挖掘結果進行充分驗證,并給出了算法在實際應用中的效果展示.后續工作將繼續研究如何對發現的大量FCP進行排序優化,從而將更符合用戶偏好的FCP排在挖掘結果的前面.