Spark機器學習(9)：FPGrowth算法

關聯規則挖掘最典型的例子是購物籃分析，通過分析可以知道哪些商品經常被一起購買，從而可以改進商品貨架的布局。

1. 基本概念

首先，介紹一些基本概念。

(1) 關聯規則：用於表示數據內隱含的關聯性，一般用X表示先決條件，Y表示關聯結果。

(2) 支持度(Support)：所有項集中{X,Y}出現的可能性。

(3) 置信度(Confidence)：先決條件X發生的條件下，關聯結果Y發生的概率。

2. Apriori算法

Apriori算法是常用的關聯規則挖掘算法，基本思想是：

(1) 先搜索出1項集及其對應的支持度，刪除低於支持度的項集，得到頻繁1項集L1；

(2) 對L1中的項集進行連接，得到一個候選集，刪除其中低於支持度的項集，得到頻繁1項集L2；

...

迭代下去，一直到無法找到L(k+1)為止，對應的頻繁k項集集合就是最后的結果。

Apriori算法的缺點是對於候選項集里面的每一項都要掃描一次數據，從而需要多次掃描數據，I/O操作多，效率低。為了提高效率，提出了一些基於Apriori的算法，比如FPGrowth算法。

3. FPGrowth算法

FPGrowth算法為了減少I/O操作，提高效率，引入了一些數據結構存儲數據，主要包括項頭表、FP-Tree和節點鏈表。

3.1 項頭表

項頭表(Header Table)即找出頻繁1項集，刪除低於支持度的項集，並按照出現的次數降序排序，這是第一次掃描數據。然后從數據中刪除非頻繁1項集，並按照項頭表的順序排序，這是第二次也是最后一次掃描數據。

下面的例子，支持度=0.4，閾值=0.4*10=4，因為D、F、G出現次數小於4次，小於閾值，所以被刪除，項頭表按照各一項集出現的次數重新排序。如ABCE=>EABC。

3.2 FP-Tree

3.2.1 FP-Tree的建立

FP-Tree(Frequent Pattern Tree)初始時只有一個根節點Null，將每一條數據里的每一項，按照排序后的順序插入FP-Tree，節點的計數為1，如果有共用的祖先，則共用祖先的節點計數+1。

首先，插入第1條數據E：

插入第2條數據ABC：

插入第3條數據EABC：

以此類推，所有數據都插入以后：

 

3.2.2 FP-Tree的挖掘過程

FP-Tree的挖掘過程如下，從長度為1的頻繁模式開始挖掘。可以分為3個步驟：

(1) 構造它的條件模式基(CPB, Conditional Pattern Base)，條件模式基(CPB)就是我們要挖掘的Item的前綴路徑；

(2) 然后構造它的條件FP-Tree(Conditional FP-tree)；

(3) 遞歸的在條件FP Tree上進行挖掘。

從項頭表的最下面一項（也就是C）開始，包含C的3個CPB分別是EAB、E、AB，其計數分別為2、1、2，可以表示為CPB{<EAB:2>,<E:1>,<AB:2>}。累加每個CPB上的Item計數，低於閾值的刪除，得到條件FP Tree(Conditional FP-tree)。如CPB{<EAB:2>,<E:1>,<AB:2>}，得到E:3，A:4，B:4，E的計數小於閾值4，所以刪除，得到C的條件FP Tree如下：

 

在條件FP Tree上使用如下的算法進行挖掘：

procedure FP_growth(Tree, α){
if Tree 含單個路徑P {
    for 路徑 P 中結點的每個組合（記作β）{
        產生模式β ∪ α，其支持度support = β中結點的最小支持度；
    }
}
else {
    for each a i 在 Tree 的頭部 {
        產生一個模式β = ai ∪ α，其支持度support = ai.support；
        構造β的條件模式基，然后構造β的條件FP Tree Treeβ；
        if Treeβ ≠ ∅ then
            調用FP_growth (Treeβ, β)；}
    }
}

 對於上面的條件FP Tree，可知是單個路徑，可以得到以下的頻繁模式：<AC:4>、<BC:4>、<ABC:4>。

4. MLlib的FPGrowth算法

直接上代碼：

import org.apache.log4j.{ Level, Logger }
import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.rdd.RDD
import org.apache.spark.mllib.fpm.{ FPGrowth, FPGrowthModel }

/**
  * Created by Administrator on 2017/7/16.
  */
object FPGrowth {

  def main(args:Array[String]) ={
    // 設置運行環境
    val conf = new SparkConf().setAppName("FPGrowth")
      .setMaster("spark://master:7077").setJars(Seq("E:\\Intellij\\Projects\\MachineLearning\\MachineLearning.jar"))
    val sc = new SparkContext(conf)
    Logger.getRootLogger.setLevel(Level.WARN)

    // 讀取樣本數據並解析
    val dataRDD = sc.textFile("hdfs://master:9000/ml/data/sample_fpgrowth.txt")
    val exampleRDD = dataRDD.map(_.split(" ")).cache()

    // 建立FPGrowth模型,最小支持度為0.4
    val minSupport = 0.4
    val numPartition = 10
    val model = new FPGrowth().
      setMinSupport(minSupport).
      setNumPartitions(numPartition).
      run(exampleRDD)

    // 輸出結果
    println(s"Number of frequent itemsets: ${model.freqItemsets.count()}")
    model.freqItemsets.collect().foreach { itemset =>
      println(itemset.items.mkString("[", ",", "]") + ":" + itemset.freq)
    }
  }

}

樣本數據：

D E
A B C
A B C E
B E
C D E
A B C
A B C E
B E
F G
D F

運行結果：

 

參考文獻：《數據挖掘概念與技術》。