Spark Mllib中关联挖掘FP-growth

1.关联规则

关联性强度，由3个概念，即支持度，置信度，提升度来控制和评价。
<1>
支持度：是指在所有项集中{X,Y}出现的可能性，即项集中同时包含X和Y的概率。假设设置最小支持度阈值为5%，由于{尿布，啤酒}的支持度为800/10000=8%，满足最小阈值要求，称为频繁项集，保留规则。
<2>
置信度：表示在先决条件X发生的条件下，关联结果Y发生的概率：这是生成强关联规则的第二个门槛，衡量了所考察的关联规则在“质”上的可靠性。
当设定置信度的最小阈值为70%，例如{啤酒，尿布}中，购买尿布时会购买啤酒的置信度为800/1000=80%，保留规则；而购买啤酒时购买尿布的概率是800/2000=40%,则被剔除。
到底是由X产生Y，还是由Y产生X。
<3>
提升度：表示在含有X条件下同时含有Y的可能性，与没有X这个条件下项集含有Y的可能性之比，即
置信度(X=>Y)/支持度(Y)

2.FPGrowth算法

2.1 构造FP树

image.png

FP-tree其实是一棵前缀树，按支持度降序排列，支持度越高的频繁项离根节点越近，从而使得更多的频繁项可以共享前缀。

第一次扫描：

| 商品 | 包含该商品的记录数 |
| I1 | 6 |
| I2 | 7 |
| I3 | 6 |
| I4 | 2 |
| I5 | 2 |

调整之后的事务数据库：

image.png

进行第二次扫描，构建FP-tree（按照支持度降序）：

(1) FP-tree的根节点为null，不表示任何项。

第一条记录{I1,I2,I5}，其中支持度最高的是I2为7次，之后依次是I1,I5

image.png

(2) 第二条记录{I2,I4}，与第一条记录有相同的前缀I2，因此<I2>的支持度加一，同时在(I2:2)节点下添加节点(I4:1)，

所以，FP-tree中的第二条分支是<(I2:2)，(I4:1)>

image.png

（3）同理，第3条记录

image.png

（4）第四条记录<I2,I1,I4>与之前所有记录共同前缀<I2,I1>，因此在<I2,I1>节点下添加节点<I4>：

image.png

（5）第5条记录{I1，I3}，当前FP-tree并没有I1前缀，所以新建新的分支：

image.png

依次类推最后的树：

image.png

2.2 从FP-tree中挖掘频繁模式

设定最小阈值为2
（1）挖掘I5

image.png
所以I5的频繁项集就是{ {I2,I5:2}，{I1,I5:2}，{I2,I1,I5:2} }

（2）挖掘I3
{I2,I1,I3} {I2,I3} {I1,I3}

image.png

I3的条件FP树仍然是一个多路径树：
a、首先将模式后缀I3和条件FP树表头项的每一项取并集，得到一组模式为{I2 I3:4，I1 I3:2}，但是这一组模式不是后缀为I3的所有模式，还需要调用Fp_growth，模式后缀为{I1，I3}，{I1，I3}的条件基为{I2：2}

image.png

3.Spark Mllib中FPGrowth算法

image.png
image.png

FPGrowth源码分解：

image.png

run方法：

  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   //数据分区设置,就是FP-growth的计算并行
    val partitioner = new HashPartitioner(numParts)
    val freqItems = genFreqItems(data, minCount, partitioner)   //计算满足最小支持度的Item项
    val freqItemsets = genFreqItemsets(data, minCount, freqItems, partitioner)   //计算频繁项集
    new FPGrowthModel(freqItemsets)
  }

获取频繁项 genFreqItems方法：

  private def genFreqItems[Item: ClassTag](
      data: RDD[Array[Item]],
      minCount: Long,
      partitioner: Partitioner): Array[Item] = {
    data.flatMap { t =>
      val uniq = t.toSet
      if (t.length != uniq.size) {
        throw new SparkException(s"Items in a transaction must be unique but got ${t.toSeq}.")
      }
      t
    }.map(v => (v, 1L))
      .reduceByKey(partitioner, _ + _)
      .filter(_._2 >= minCount)
      .collect()
      .sortBy(-_._2)
      .map(_._1)
  }

获取频繁项集 genFreqItemsets方法

  private def genFreqItemsets[Item: ClassTag](
      data: RDD[Array[Item]],
      minCount: Long,
      freqItems: Array[Item],
      partitioner: Partitioner): RDD[FreqItemset[Item]] = {
    val itemToRank = freqItems.zipWithIndex.toMap   // 频繁项 --> rank
    data.flatMap { transaction =>
      genCondTransactions(transaction, itemToRank, partitioner)   // 每一个事务中生成的分区事务集
    }.aggregateByKey(new FPTree[Int], partitioner.numPartitions)(  //新建FP树,进行分区合并
      (tree, transaction) => tree.add(transaction, 1L),
      (tree1, tree2) => tree1.merge(tree2))   //树合并
    .flatMap { case (part, tree) =>
      tree.extract(minCount, x => partitioner.getPartition(x) == part)  //抽取频繁项集
    }.map { case (ranks, count) =>
      new FreqItemset(ranks.map(i => freqItems(i)).toArray, count)  // 返回频繁项集
    }
  }