Spark MLlib FPGrowth算法

1.1 FPGrowth算法

1.1.1 基本概念

关联规则挖掘的一个典型例子是购物篮分析。关联规则研究有助于发现交易数据库中不同商品（项）之间的联系，找出顾客购买行为模式，如购买了某一商品对购买其他商品的影响，分析结果可以应用于商品货架布局、货存安排以及根据购买模式对用户进行分类。

关联规则的相关术语如下：

（1）项与项集

这是一个集合的概念，在一篮子商品中的一件消费品即为一项（Item），则若干项的集合为项集，如{啤酒，尿布}构成一个二元项集。

（2）关联规则

一般记为的形式，X为先决条件，Y为相应的关联结果，用于表示数据内隐含的关联性。如：表示购买了尿布的消费者往往也会购买啤酒。

关联性强度如何，由三个概念——支持度、置信度、提升度来控制和评价。

例：有10000个消费者购买了商品，其中购买尿布1000个，购买啤酒2000个，购买面包500个，同时购买尿布和面包800个，同时购买尿布和面包100个。

（3）支持度（Support）

支持度是指在所有项集中{X, Y}出现的可能性，即项集中同时含有X和Y的概率：

该指标作为建立强关联规则的第一个门槛，衡量了所考察关联规则在“量”上的多少。通过设定最小阈值（minsup），剔除“出镜率”较低的无意义规则，保留出现较为频繁的项集所隐含的规则。

设定最小阈值为5%，由于{尿布，啤酒}的支持度为800/10000=8%，满足基本输了要求，成为频繁项集，保留规则；而{尿布，面包}的支持度为100/10000=1%，被剔除。

（4）置信度（Confidence）

置信度表示在先决条件X发生的条件下，关联结果Y发生的概率：

这是生成强关联规则的第二个门槛，衡量了所考察的关联规则在“质”上的可靠性。相似的，我们需要对置信度设定最小阈值（mincon）来实现进一步筛选。

具体的，当设定置信度的最小阈值为70%时，置信度为800/1000=80%，而的置信度为800/2000=40%，被剔除。

（5）提升度（lift）

提升度表示在含有X的条件下同时含有Y的可能性与没有X这个条件下项集中含有Y的可能性之比：公式为confidence(artichok => cracker)/support(cracker) = 80%/50% = 1.6。该指标与置信度同样衡量规则的可靠性，可以看作是置信度的一种互补指标。

1.1.2 FP-Growth算法

FP-Growth(频繁模式增长)算法是韩家炜老师在2000年提出的关联分析算法，它采取如下分治策略：将提供频繁项集的数据库压缩到一棵频繁模式树（FP-Tree），但仍保留项集关联信息；该算法和Apriori算法最大的不同有两点：第一，不产生候选集，第二，只需要两次遍历数据库，大大提高了效率。

（1）按以下步骤构造FP-树

(a) 扫描事务数据库D一次。收集频繁项的集合F和它们的支持度。对F按支持度降序排序，结果为频繁项表L。

(b) 创建FP-树的根结点，以“null”标记它。对于D 中每个事务Trans，执行：选择 Trans 中的频繁项，并按L中的次序排序。设排序后的频繁项表为[p | P]，其中，p 是第一个元素，而P 是剩余元素的表。调用insert_tree([p | P], T)。该过程执行情况如下。如果T有子女N使得N.item-name = p.item-name，则N 的计数增加1；否则创建一个新结点N将其计数设置为1，链接到它的父结点T，并且通过结点链结构将其链接到具有相同item-name的结点。如果P非空，递归地调用insert_tree(P, N)。

（2）FP-树的挖掘

通过调用FP_growth(FP_tree, null)实现。该过程实现如下：

FP_growth(Tree, α)

(1) if Tree 含单个路径P then

(2) for 路径 P 中结点的每个组合（记作β）

(3) 产生模式β ∪ α，其支持度support = β中结点的最小支持度；

(4) else for each ai在Tree的头部(按照支持度由低到高顺序进行扫描) {

(5) 产生一个模式β = ai ∪ α，其支持度support = ai .support；

(6) 构造β的条件模式基，然后构造β的条件FP-树Treeβ；

(7) if Treeβ ≠ ∅ then

(8) 调用 FP_growth (Treeβ, β)；}

end

1.1.3 FP-Growth算法演示—构造FP-树

（1）事务数据库建立

原始事务数据库如下：

Tid	Items
1	I1,I2,I5
2	I2,I4
3	I2,I3
4	I1,I2,I4
5	I1,I3
6	I2,I3
7	I1,I3
8	I1,I2,I3,I5
9	I1,I2,I3

扫描事务数据库得到频繁1-项目集F。

I1	I2	I3	I4	I5
6	7	6	2	2

定义minsup=20%，即最小支持度为2，重新排列F。

I2	I1	I3	I4	I5
7	6	6	2	2

重新调整事务数据库。

Tid	Items
1	I2, I1,I5
2	I2,I4
3	I2,I3
4	I2, I1,I4
5	I1,I3
6	I2,I3
7	I1,I3
8	I2, I1,I3,I5
9	I2, I1,I3

（2）创建根结点和频繁项目表

（3）加入第一个事务(I2,I1,I5)

（4）加入第二个事务(I2,I4)

（5）加入第三个事务(I2,I3)

以此类推加入第5、6、7、8、9个事务。

（6）加入第九个事务(I2,I1,I3)

1.1.4 FP-Growth算法演示—FP-树挖掘

FP-树建好后，就可以进行频繁项集的挖掘，挖掘算法称为FpGrowth（Frequent Pattern Growth）算法，挖掘从表头header的最后一个项开始，以此类推。本文以I5、I3为例进行挖掘。

（1）挖掘I5：

对于I5，得到条件模式基：<(I2,I1:1)>、<I2,I1,I3:1>

构造条件FP-tree：

得到I5频繁项集：{{I2,I5:2},{I1,I5:2},{I2,I1,I5:2}}

I4、I1的挖掘与I5类似，条件FP-树都是单路径。

（1）挖掘I3：

I5的情况是比较简单的，因为I5对应的条件FP-树是单路径的，I3稍微复杂一点。I3的条件模式基是(I2 I1:2), (I2:2), (I1:2)，生成的条件FP-树如下图：

I3的条件FP-树仍然是一个多路径树，首先把模式后缀I3和条件FP-树中的项头表中的每一项取并集，得到一组模式{I2 I3:4, I1 I3:4}，但是这一组模式不是后缀为I3的所有模式。还需要递归调用FP-growth，模式后缀为{I1，I3}，{I1，I3}的条件模式基为{I2：2}，其生成的条件FP-树如下图所示。

在FP_growth中把I2和模式后缀{I1，I3}取并得到模式{I1 I2 I3：2}。

理论上还应该计算一下模式后缀为{I2，I3}的模式集，但是{I2，I3}的条件模式基为空，递归调用结束。最终模式后缀I3的支持度>2的所有模式为：{ I2 I3:4, I1 I3:4, I1 I2 I3:2}。

1.2 Spark Mllib FPGrowth源码分析

FPGrowth源码包括：FPGrowth、FPTree两部分。

其中FPGrowth中包括：run方法、genFreqItems方法、genFreqItemsets方法、genCondTransactions方法；

FPTree中包括：add方法、merge方法、project方法、getTransactions方法、extract方法。

// run 计算频繁项集

/**

* Computes an FP-Growth model that contains frequent itemsets.

* @param data input data set, each element contains a transaction

* @return an [[FPGrowthModel]]

def run[Item: ClassTag](data: RDD[Array[Item]]): FPGrowthModel[Item] = {

if (data.getStorageLevel == StorageLevel.NONE) {

logWarning("Input data is not cached.")

}

val count = data.count()//计算事务总数

val minCount = math.ceil(minSupport * count).toLong//计算最小支持度

val numParts = if (numPartitions > 0) numPartitions else data.partitions.length

val partitioner = new HashPartitioner(numParts)

//freqItems计算满足最小支持度的Items项

val freqItems = genFreqItems(data, minCount, partitioner)

//freqItemsets计算频繁项集

val freqItemsets = genFreqItemsets(data, minCount, freqItems, partitioner)

new FPGrowthModel(freqItemsets)

}

// genFreqItems计算满足最小支持度的Items项

/**

* Generates frequent items by filtering the input data using minimal support level.

* @param minCount minimum count for frequent itemsets

* @param partitioner partitioner used to distribute items

* @return array of frequent pattern ordered by their frequencies

privatedef genFreqItems[Item: ClassTag](

data: RDD[Array[Item]],

minCount: Long,

partitioner: Partitioner): Array[Item] = {

data.flatMap { t =>

val uniq = t.toSet

if (t.size != uniq.size) {

thrownew SparkException(s"Items in a transaction must be unique but got ${t.toSeq}.")

}

}.map(v => (v, 1L))

.reduceByKey(partitioner, _ + _)

.filter(_._2 >= minCount)

.collect()

.sortBy(-_._2)

.map(_._1)

}//统计每个Items项的频次，对小于minCount的Items项过滤，返回Items项。

// genFreqItemsets计算频繁项集：生成FP-Trees，挖掘FP-Trees

/**

* Generate frequent itemsets by building FP-Trees, the extraction is done on each partition.

* @param data transactions

* @param minCount minimum count for frequent itemsets

* @param freqItems frequent items

* @param partitioner partitioner used to distribute transactions

* @return an RDD of (frequent itemset, count)

privatedef genFreqItemsets[Item: ClassTag](

data: RDD[Array[Item]],

minCount: Long,

freqItems: Array[Item],

partitioner: Partitioner): RDD[FreqItemset[Item]] = {

val itemToRank = freqItems.zipWithIndex.toMap//表头

data.flatMap { transaction =>

genCondTransactions(transaction, itemToRank, partitioner)

}.aggregateByKey(new FPTree[Int], partitioner.numPartitions)( //生成FP树

(tree, transaction) => tree.add(transaction, 1L), //FP树增加一条事务

(tree1, tree2) => tree1.merge(tree2)) //FP树合并

.flatMap { case (part, tree) =>

tree.extract(minCount, x => partitioner.getPartition(x) == part)//FP树挖掘频繁项

}.map { case (ranks, count) =>

new FreqItemset(ranks.map(i => freqItems(i)).toArray, count)

}

// add FP-Trees增加一条事务数据

/** Adds a transaction with count. */

def add(t: Iterable[T], count: Long = 1L): this.type = {

require(count > 0)

var curr = root

curr.count += count

t.foreach { item =>

val summary = summaries.getOrElseUpdate(item, new Summary)

summary.count += count

val child = curr.children.getOrElseUpdate(item, {

val newNode = new Node(curr)

newNode.item = item

summary.nodes += newNode

newNode

})

child.count += count

curr = child

}

this

}

// merge FP-Trees合并

/** Merges another FP-Tree. */

def merge(other: FPTree[T]): this.type = {

other.transactions.foreach { case (t, c) =>

add(t, c)

}

this

}

// extract FP-Trees挖掘，返回所有频繁项集

/** Extracts all patterns with valid suffix and minimum count. */

def extract(

minCount: Long,

validateSuffix: T => Boolean = _ => true): Iterator[(List[T], Long)] = {

summaries.iterator.flatMap { case (item, summary) =>

if (validateSuffix(item) && summary.count >= minCount) {

Iterator.single((item :: Nil, summary.count)) ++

project(item).extract(minCount).map { case (t, c) =>

(item :: t, c)

}

} else {

Iterator.empty

}

1.3 Mllib FPGrowth实例

1、数据

数据格式为：物品1物品2物品3…

r z h k p

z y x w v u t s

s x o n r

x z y m t s q e

x z y r q t p

2、代码

//读取样本数据

valdata_path = "/home/tmp/sample_fpgrowth.txt"

valdata = sc.textFile(data_path)

valexamples = data.map(_.split(" ")).cache()

//建立模型

valminSupport = 2

valnumPartition = 10

valmodel = new FPGrowth()

.setMinSupport(minSupport)

.setNumPartitions(numPartition)

.run(examples)

//打印结果

println(s"Number of frequent itemsets: ${model.freqItemsets.count()}")

model.freqItemsets.collect().foreach { itemset =>

println(itemset.items.mkString("[", ",", "]") + ", " + itemset.freq)

}