一種基于Flink的流關聯挖掘算法

馮 鵬，胡佳麗，黃 山,裴騰達，段曉東

(1.大連大學 大連市環境感知與智能控制重點實驗室，遼寧 大連 116622；2.大連民族大學a.計算機科學與工程學院；b.大數據應用技術國家民委重點實驗室，遼寧 大連 116650)

5G及互聯網技術與各行業相結合，使許多行業產生高速實時的流式數據。基于流式數據的關聯規則挖掘，可以發現事務間的隱藏聯系，具有廣泛的應用范圍。流數據具有大量連續隨機出現、體量潛在無限大、不斷出現、不能控制數據流項目序列的到來順序等特點。因此，流數據在挖掘處理時需要追蹤更多的信息，如之前的頻繁項在一段時間后可能轉為非頻繁項，亦或者之前的非頻繁項一段時間后轉為頻繁項。而且數據不斷的輸入，使得系統內存中的數據需要不斷維護與調整。在此背景下，一些經典的挖掘算法如Apriori[1]、FP-Growth[2]及CHARM[3]等，已無法維持流數據增量式更新的要求。Chris G等人[4]針對流數據挖掘過程中的時間敏感性，提出了流數據挖掘算法FP-Stream，但難以滿足現階段流式大數據關聯規則挖掘的需求。Storm[5]、Spark-streaming[6]、Flink[7]等流框架的出現，為大量經典的關聯規則挖掘算法提供了新的生機。在此背景下，本文提出一種基于Flink的流關聯挖掘算法FP-Flink。

1 背景知識

首先介紹本文提出的FP-Flink算法的兩個基本結構，全局模式樹(Pattern-Tree)與傾斜時間窗口。然后介紹相關基本概念。

(1)全局模式樹。全局頻繁模式樹是現階段內所有頻繁模式的統一表達。FP-Flink把FP樹中的支持度計數改為滑動時間窗口表便形成了其特有的FP-Flink結構如圖1。表中記錄著該頻繁模式從過去到當下時間窗口內每階段的頻率。通過這種記錄方式詳細記錄近期的數據，粗略記錄時間較為久遠的數據，來保證在跨度較大的時間段內對數據信息進行保存，從而緩解計算機的存儲壓力。

圖1 帶有時間窗口的FP樹

(2)傾斜時間窗口。在數據挖掘過程中，不僅要關注頻繁項集，還必須注意對潛在頻繁項集進行維護，因為潛在頻繁項集可能隨著時間變化轉化為頻繁項集，而無法轉變的非頻繁項集則應該被丟棄。傾斜時間窗口的基本思想為：相較于久遠的信息，人們對當下的信息更感興趣。以本文所使用的對數傾斜時間窗口，先選擇一個最小時間粒度T作為當前整個的時間窗口的基礎，當然T可以任意取值，然后以2為對數進行求值即可，在此模型下，若想求得最近一年的數據，同樣取一分鐘為最小時間粒度，則與之相鄰近的2 min、4 min及8 min……均需進行處理，則一年數據的對數傾斜時間窗口模型如圖2。

圖2 對數傾斜時間窗口模型

(3)相關基本概念。 設I={i1，i2，…，ik，…，im}是一系列項的集合，則項集X可以定義為X={i1，i2，…，ik}?I是一個k項集，事務T可以定義為T=(Tid，X)，其中Tid被稱作事務標識。以此類推，我們可以定義某事務集D={T1，T2，…，Tk，…，Tn}。則稱D中所包含的項集Z的事務數量Sup(Z)為項集Z的支持度。最小支持度minsup(0

設松弛變量ρ=ε/minsup，顯然可以根據ρ得到最大支持度誤差ε，則以此可以把項集分為三種情況：(1)若Sup(X)≥σ，則項集Z為D中的頻繁項集；(2)若ε≤ Sup(Z) < minsup，則Z為D中的潛在頻繁項集；(3)若Sup(Z)<ε，則Z為D中的非頻繁項集。

2 基于Flink的流關聯挖掘算法

雖然FP-Flink算法所采用的樹結構在一定程度上也能夠緩解內存壓力，但在高速的流數據面前，FP-Flink結構會異常龐大，遍歷樹結構完成挖掘任務也會更加費時，不能滿足用戶對實時性的要求。FP-Flink算法的分布式模型如圖3。FP-Flink先把數據劃分配給不同的節點，然后在所有節點記錄數據概要信息，接著挖掘投影前綴子樹生成初步的FP-Flink結構，最后把FP-Flink結構更新匯總。

圖3 FP-Flink分布式模型

FP-Flink算法的核心思想是采用劃分投影、字典序結構及序列化存儲的方式，降低算法計算過程對空間的過度占用問題，并加快數據的挖掘效率。FP-Flink算法步驟主要有：(1)數據的預處理，主要包括數據的讀入與頻繁1項集的生成；(2)剔除非頻繁項集；(3)劃分投影，即把大數據集分成小數據集；(4)挖掘投影子樹，即生成初步的FP-Flink結構，所生成的FP-Flink初步結構為字典序結構；(5)更新FP-Flink樹結構，包括把FP-Flink序列化存入Redis與全局更新時將其取出更新兩個操作。

(1)數據預處理。本操作主要是對Flink讀入的數據做一個簡單的詞頻統計，得到其中的潛在頻繁一項集，以方便下一步剔除非潛在頻繁項。即通過輸入數據集和最小潛在支持度，輸出中一切頻率大于的項的集合。首先，Flink接收來自數據源的一個批次的數據，即對于事務，過濾其中的每一項，然后輸出元組，綜合得到的頻率以及所有項的總計數，當的時候，則認為為頻繁1項集，并輸出。數據預處理偽代碼見表1。

表1 FP-Flink數據預處理偽代碼

(2)剔除非潛在頻繁項。本操作將剔除當前時刻批次內數據事務T中非潛在頻繁的項，并按照字典序[8]的偏序關系對頻繁項集進行排序。以表2中的數據為例，設最小支持度為2，則經過數據預處理和剔除非頻繁項這兩步，得到的結果見表3。

表2 某批次數據集

(3)劃分投影。設定劃分規則，盡量保證把上一步產生的頻繁1項集中的所有項能夠均衡地分配給不同的節點。根據設定好的劃分規則，把數據分配到不同的計算節點之中。針對某一條事務，從該事務的最后一項開始，逐項向該事務的第一項進行遍歷，并根據當前最后一項的值進行劃分，把其第一項到該事務集的所有項，分配給劃分規則中對應的節點。當對應節點已經存有該條子記錄時，則跳過這一項。直到該條事務的最小的子記錄也存在于計算節點中，則結束遍歷。假設存在一個具有3個節點的集群，則表3中的數據，按照Hash取余的劃分投影方式，其數據分配結果見表4。

表3 排序結果

表4 數據劃分結果

這種數據劃分方法的使用，使得劃分后的數據集同時進行運算挖掘成為了可能，消除了節點間不必要的數據通信。同時降低了一些節點里邊子記錄的重復率，緩解了計算時的內存壓力。

(4)挖掘投影子樹。本步驟主要是生成FP-Flink的初步結構。每個節點在接收到相應的數據后，會把其對應的消息添加到其自己的子樹中，每個投影子樹只記錄著全局前綴樹的一部分數據。按照上一步的劃分投影規則，每個計算節點中事務的最后一項一定是劃分規則中的某一項。因此挖掘投影子樹的時候，本文選擇從每個事務的最后一項為起始項，采用自下而上的方式遍歷挖掘。以表4為例，則其節點2可以得到挖掘結果如圖4。

圖4 節點2的FP-Flink初步結構

(5)FP-Flink樹更新。這一步包括兩個操作，首先需要將初步生成的FP-Flink結構序列化成字節流后存入Redis中，這樣可以大大減少FP-Flink結構對內存的占用。最終更新FP-Flink的全局模式時，可以根據需要以數據讀取批次數、或者固定時間間隔進行FP-Flink結構的全局整合，如系統每讀取5批次的數據或每間隔500 ms，便進行一次FP-Flink結構更新，更新算法見表5。

表5 FP-Flink全局模式樹更新算法

3 實驗測評

3.1 實驗環境配置

實驗在三臺機器所組成的集群上進行，采用一主兩從結構，實驗相關配置見表6。

表6 實驗環境配置

3.2 實驗結果與分析

本文使用合成數據集評測FP-Flink算法性能。由于流數據本身具有很大的不確定性及高并發的特點，為了降低數據源對實驗的影響，同時保證流數據的特性，本文選擇Kafka作為消息中間件接收各個渠道的數據流。實驗從算法的精確性和運行時間兩個角度，對FP-Flink算法與經典的FP-Stream算法進行比較。

(1)測試頻繁項集隨支持度變化情況。隨著最小支持度的減小，FP-Flink和FP-Stream算法的挖掘時間都會以指數級爆炸增長如圖5。因為隨著最小支持度的不斷減小，整個頻繁1項集會大幅增加，從而生成更為龐大的頻繁模式樹，加劇算法挖掘時的耗時。當最小支持度處于0.7到0.95的區間內，頻繁項集較少，FP-Flink算法因為有著額外的工作節點間調度和數據通信會造成的額外開銷，其算法運行時間會比FP-Stream算法稍微高一點點。當最小支持度小于0.7時，隨著挖掘出的頻繁項集的增多，因為FP-Flink算法所有的計算節點的頻繁模式樹和全局頻繁模式樹均小于FP-Stream算法的，故而其時間優勢也越來越明顯。

圖5 最小支持度對算法影響

(2)吐量測評。隨著系統的吞吐量的不斷增大，FP-Stream算法的運行時間增長率逐減增大后慢慢趨于平穩如圖6。FP-Flink的運行時間也隨著系統吞吐量的增大不斷增大，但整個測試時間段內，FP-Flink算法運行時間均優于FP-Stream算法。

圖6 數據吞吐量對運行時間的影響

(3)可伸縮性測試。在數據集和支持度相同的條件下時，FP-stream與FP-Flink算法的挖掘結果相同。本文提出的FP-Flink算法隨著數據集的增大有著較好的線性可伸縮性，而經典的FP-stream算法則沒有，算法可拓展性測試如圖7。FP-Flink算法運行的時候，其每個線程都會生成比FP-Stream更小的樹，所以對樹進行遍歷所需要的時間也更少，故而FP-Flink運行時間比FP-Stream少。

圖7 算法可拓展性測試

4 總結與展望

本文提出一種基于Flink的流關聯挖掘算法FP-Flink，其核心思想是通過劃分投影、字典序結構與序列化存儲的方式，解決流關聯挖掘過程中，數據中間結果太大，過度占用內存的問題。經實驗驗證，該算法可以緩解計算時的內存壓力，縮減數據的挖掘時間，相較于經典流挖掘算法具有更快的處理速度且具有更強的可擴展性。受研究環境有限，本文僅研究了具有萬條數據的流關聯挖掘。以后需要在更大的數據集及更多流數據上應用，以進一步驗證算法的有效性。