频繁模式挖掘 - 基于 RDD 的 API

挖掘频繁项、项集、子序列或其他子结构通常是分析大规模数据集的首要步骤之一，多年来一直是数据挖掘领域的一个活跃的研究课题。我们建议用户参考维基百科的关联规则学习获取更多信息。spark.mllib 提供了 FP-growth 的并行实现，FP-growth 是一种流行的频繁项集挖掘算法。

FP-growth

FP-growth 算法在论文 Han et al., Mining frequent patterns without candidate generation 中进行了描述，其中“FP”代表频繁模式。给定一个事务数据集，FP-growth 的第一步是计算项目频率并识别频繁项目。与为相同目的设计的 Apriori-like 算法不同，FP-growth 的第二步使用后缀树 (FP-tree) 结构来编码事务，而无需显式生成候选集，而候选集通常生成成本很高。第二步之后，可以从 FP-树中提取频繁项集。在 spark.mllib 中，我们实现了 FP-growth 的并行版本，称为 PFP，如 Li et al., PFP: Parallel FP-growth for query recommendation 中所述。 PFP 基于事务的后缀来分发 FP-树增长的工作，因此比单机实现更具可扩展性。我们建议用户参阅论文以获取更多详细信息。

spark.mllib 的 FP-growth 实现采用以下（超）参数

minSupport: 一个项集被识别为频繁项集所需的最小支持度。例如，如果一个项目在 5 个事务中出现 3 次，则其支持度为 3/5=0.6。
numPartitions：用于分配工作的分区数。

示例

FPGrowth 实现了 FP-growth 算法。它接受一个事务的 RDD，其中每个事务都是一个通用类型的项目 List。使用事务调用 FPGrowth.train 会返回一个 FPGrowthModel，该模型存储频繁项集及其频率。

有关 API 的更多详细信息，请参阅 FPGrowth Python 文档。

from pyspark.mllib.fpm import FPGrowth

data = sc.textFile("data/mllib/sample_fpgrowth.txt")
transactions = data.map(lambda line: line.strip().split(' '))
model = FPGrowth.train(transactions, minSupport=0.2, numPartitions=10)
result = model.freqItemsets().collect()
for fi in result:
    print(fi)

在 Spark 仓库的 "examples/src/main/python/mllib/fpgrowth_example.py" 中查找完整的示例代码。

FPGrowth 实现了 FP-growth 算法。它接受一个事务的 RDD，其中每个事务都是一个通用类型的项目 Array。使用事务调用 FPGrowth.run 会返回一个 FPGrowthModel，该模型存储频繁项集及其频率。以下示例说明了如何从 transactions 中挖掘频繁项集和关联规则（有关详细信息，请参阅关联规则）。

有关 API 的详细信息，请参阅 FPGrowth Scala 文档。

import org.apache.spark.mllib.fpm.FPGrowth
import org.apache.spark.rdd.RDD

val data = sc.textFile("data/mllib/sample_fpgrowth.txt")

val transactions: RDD[Array[String]] = data.map(s => s.trim.split(' '))

val fpg = new FPGrowth()
  .setMinSupport(0.2)
  .setNumPartitions(10)
val model = fpg.run(transactions)

model.freqItemsets.collect().foreach { itemset =>
  println(s"${itemset.items.mkString("[", ",", "]")},${itemset.freq}")
}

val minConfidence = 0.8
model.generateAssociationRules(minConfidence).collect().foreach { rule =>
  println(s"${rule.antecedent.mkString("[", ",", "]")}=> " +
    s"${rule.consequent .mkString("[", ",", "]")},${rule.confidence}")
}

在 Spark 仓库的 "examples/src/main/scala/org/apache/spark/examples/mllib/SimpleFPGrowth.scala" 中查找完整的示例代码。

FPGrowth 实现了 FP-growth 算法。它接受一个事务的 JavaRDD，其中每个事务都是一个通用类型的项目 Iterable。使用事务调用 FPGrowth.run 会返回一个 FPGrowthModel，该模型存储频繁项集及其频率。以下示例说明了如何从 transactions 中挖掘频繁项集和关联规则（有关详细信息，请参阅关联规则）。

有关 API 的详细信息，请参阅 FPGrowth Java 文档。

import java.util.Arrays;
import java.util.List;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.fpm.AssociationRules;
import org.apache.spark.mllib.fpm.FPGrowth;
import org.apache.spark.mllib.fpm.FPGrowthModel;

JavaRDD<String> data = sc.textFile("data/mllib/sample_fpgrowth.txt");

JavaRDD<List<String>> transactions = data.map(line -> Arrays.asList(line.split(" ")));

FPGrowth fpg = new FPGrowth()
  .setMinSupport(0.2)
  .setNumPartitions(10);
FPGrowthModel<String> model = fpg.run(transactions);

for (FPGrowth.FreqItemset<String> itemset: model.freqItemsets().toJavaRDD().collect()) {
  System.out.println("[" + itemset.javaItems() + "], " + itemset.freq());
}

double minConfidence = 0.8;
for (AssociationRules.Rule<String> rule
  : model.generateAssociationRules(minConfidence).toJavaRDD().collect()) {
  System.out.println(
    rule.javaAntecedent() + " => " + rule.javaConsequent() + ", " + rule.confidence());
}

在 Spark 仓库的 "examples/src/main/java/org/apache/spark/examples/mllib/JavaSimpleFPGrowth.java" 中查找完整的示例代码。

关联规则

AssociationRules 实现了一种并行规则生成算法，用于构造以单个项目作为结果的规则。

有关 API 的详细信息，请参阅 AssociationRules Scala 文档。

import org.apache.spark.mllib.fpm.AssociationRules
import org.apache.spark.mllib.fpm.FPGrowth.FreqItemset

val freqItemsets = sc.parallelize(Seq(
  new FreqItemset(Array("a"), 15L),
  new FreqItemset(Array("b"), 35L),
  new FreqItemset(Array("a", "b"), 12L)
))

val ar = new AssociationRules()
  .setMinConfidence(0.8)
val results = ar.run(freqItemsets)

results.collect().foreach { rule =>
println(s"[${rule.antecedent.mkString(",")}=>${rule.consequent.mkString(",")} ]" +
    s" ${rule.confidence}")
}

在 Spark 仓库的 "examples/src/main/scala/org/apache/spark/examples/mllib/AssociationRulesExample.scala" 中查找完整的示例代码。

AssociationRules 实现了一种并行规则生成算法，用于构造以单个项目作为结果的规则。

有关 API 的详细信息，请参阅 AssociationRules Java 文档。

import java.util.Arrays;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.fpm.AssociationRules;
import org.apache.spark.mllib.fpm.FPGrowth;
import org.apache.spark.mllib.fpm.FPGrowth.FreqItemset;

JavaRDD<FPGrowth.FreqItemset<String>> freqItemsets = sc.parallelize(Arrays.asList(
  new FreqItemset<>(new String[] {"a"}, 15L),
  new FreqItemset<>(new String[] {"b"}, 35L),
  new FreqItemset<>(new String[] {"a", "b"}, 12L)
));

AssociationRules arules = new AssociationRules()
  .setMinConfidence(0.8);
JavaRDD<AssociationRules.Rule<String>> results = arules.run(freqItemsets);

for (AssociationRules.Rule<String> rule : results.collect()) {
  System.out.println(
    rule.javaAntecedent() + " => " + rule.javaConsequent() + ", " + rule.confidence());
}

在 Spark 仓库的 "examples/src/main/java/org/apache/spark/examples/mllib/JavaAssociationRulesExample.java" 中查找完整的示例代码。

PrefixSpan

PrefixSpan 是一种顺序模式挖掘算法，在 Pei et al., Mining Sequential Patterns by Pattern-Growth: The PrefixSpan Approach 中进行了描述。我们建议读者参考引用的论文，以形式化顺序模式挖掘问题。

spark.mllib 的 PrefixSpan 实现采用以下参数

minSupport：被认为是频繁顺序模式所需的最小支持度。
maxPatternLength：频繁顺序模式的最大长度。任何超过此长度的频繁模式都不会包含在结果中。
maxLocalProjDBSize：在开始对投影数据库进行本地迭代处理之前，投影数据库中允许的最大项目数。应根据执行程序的大小调整此参数。

示例

以下示例说明了在序列上运行的 PrefixSpan（使用与 Pei et al 相同的表示法）

  <(12)3>
  <1(32)(12)>
  <(12)5>
  <6>

PrefixSpan 实现了 PrefixSpan 算法。调用 PrefixSpan.run 会返回一个 PrefixSpanModel，该模型存储频繁序列及其频率。

有关 API 的详细信息，请参阅 PrefixSpan Scala 文档和 PrefixSpanModel Scala 文档。

import org.apache.spark.mllib.fpm.PrefixSpan

val sequences = sc.parallelize(Seq(
  Array(Array(1, 2), Array(3)),
  Array(Array(1), Array(3, 2), Array(1, 2)),
  Array(Array(1, 2), Array(5)),
  Array(Array(6))
), 2).cache()
val prefixSpan = new PrefixSpan()
  .setMinSupport(0.5)
  .setMaxPatternLength(5)
val model = prefixSpan.run(sequences)
model.freqSequences.collect().foreach { freqSequence =>
  println(
    s"${freqSequence.sequence.map(_.mkString("[", ", ", "]")).mkString("[", ", ", "]")}," +
      s" ${freqSequence.freq}")
}

在 Spark 仓库的 "examples/src/main/scala/org/apache/spark/examples/mllib/PrefixSpanExample.scala" 中查找完整的示例代码。

PrefixSpan 实现了 PrefixSpan 算法。调用 PrefixSpan.run 会返回一个 PrefixSpanModel，该模型存储频繁序列及其频率。

有关 API 的详细信息，请参阅 PrefixSpan Java 文档和 PrefixSpanModel Java 文档。

import java.util.Arrays;
import java.util.List;

import org.apache.spark.mllib.fpm.PrefixSpan;
import org.apache.spark.mllib.fpm.PrefixSpanModel;

JavaRDD<List<List<Integer>>> sequences = sc.parallelize(Arrays.asList(
  Arrays.asList(Arrays.asList(1, 2), Arrays.asList(3)),
  Arrays.asList(Arrays.asList(1), Arrays.asList(3, 2), Arrays.asList(1, 2)),
  Arrays.asList(Arrays.asList(1, 2), Arrays.asList(5)),
  Arrays.asList(Arrays.asList(6))
), 2);
PrefixSpan prefixSpan = new PrefixSpan()
  .setMinSupport(0.5)
  .setMaxPatternLength(5);
PrefixSpanModel<Integer> model = prefixSpan.run(sequences);
for (PrefixSpan.FreqSequence<Integer> freqSeq: model.freqSequences().toJavaRDD().collect()) {
  System.out.println(freqSeq.javaSequence() + ", " + freqSeq.freq());
}

在 Spark 仓库的 "examples/src/main/java/org/apache/spark/examples/mllib/JavaPrefixSpanExample.java" 中查找完整的示例代码。

MLlib：主要指南

MLlib：基于 RDD 的 API 指南

频繁模式挖掘 - 基于 RDD 的 API

FP-growth

关联规则

PrefixSpan