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，能夠表示爲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

運行結果：

 

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