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

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

FP-growth

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

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 等人，Mining Sequential Patterns by Pattern-Growth: The PrefixSpan Approach 中有描述。我们建议读者参考引用的论文来形式化顺序模式挖掘问题。

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

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

示例

以下示例说明了 PrefixSpan 在序列上的运行（使用与 Pei 等人相同的符号）

  <(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