频繁模式挖掘

挖掘频繁项、项集、子序列或其他子结构通常是分析大规模数据集的首要步骤之一，多年来一直是数据挖掘领域的热门研究课题。我们建议用户参考维基百科的关联规则学习以获取更多信息。

FP-Growth
PrefixSpan

FP-Growth

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

FP-growth 对项集进行操作。项集是唯一项的无序集合。 Spark 没有集合类型，因此项集表示为数组。

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

minSupport：项集被识别为频繁项所需的最小支持度。例如，如果一个项在 5 个事务中出现 3 次，则其支持度为 3/5=0.6。
minConfidence：生成关联规则的最小置信度。置信度表明关联规则被发现为真的频率。例如，如果在事务中项集 X 出现 4 次，X 和 Y 仅共同出现 2 次，则规则 X => Y 的置信度为 2/4 = 0.5。该参数不会影响频繁项集的挖掘，但会指定从频繁项集生成关联规则的最小置信度。
numPartitions：用于分配工作的分区数。默认情况下，该参数未设置，并且使用输入数据集的分区数。

FPGrowthModel 提供

freqItemsets：格式为 DataFrame 的频繁项集，其中包含以下列
- items: array：给定的项集。
- freq: long：给定配置的模型参数，此项集出现的次数。
associationRules：以 DataFrame 格式生成的置信度高于 minConfidence 的关联规则，其中包含以下列
- antecedent: array：作为关联规则假设的项集。
- consequent: array：始终包含代表关联规则结论的单个元素的项集。
- confidence: double：有关 confidence 的定义，请参阅上面的 minConfidence。
- lift: double：衡量先行词预测后继词效果的指标，计算公式为 support(antecedent U consequent) / (support(antecedent) x support(consequent))
- support: double：有关 support 的定义，请参阅上面的 minSupport。
transform：对于 itemsCol 中的每个事务，transform 方法会将其项与每个关联规则的先行词进行比较。如果记录包含特定关联规则的所有先行词，则该规则将被视为适用，并且其后继词将被添加到预测结果中。 transform 方法将总结所有适用规则中的后继词作为预测。预测列的数据类型与 itemsCol 相同，并且不包含 itemsCol 中的现有项。

示例

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

from pyspark.ml.fpm import FPGrowth

df = spark.createDataFrame([
    (0, [1, 2, 5]),
    (1, [1, 2, 3, 5]),
    (2, [1, 2])
], ["id", "items"])

fpGrowth = FPGrowth(itemsCol="items", minSupport=0.5, minConfidence=0.6)
model = fpGrowth.fit(df)

# Display frequent itemsets.
model.freqItemsets.show()

# Display generated association rules.
model.associationRules.show()

# transform examines the input items against all the association rules and summarize the
# consequents as prediction
model.transform(df).show()

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

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

import org.apache.spark.ml.fpm.FPGrowth

val dataset = spark.createDataset(Seq(
  "1 2 5",
  "1 2 3 5",
  "1 2")
).map(t => t.split(" ")).toDF("items")

val fpgrowth = new FPGrowth().setItemsCol("items").setMinSupport(0.5).setMinConfidence(0.6)
val model = fpgrowth.fit(dataset)

// Display frequent itemsets.
model.freqItemsets.show()

// Display generated association rules.
model.associationRules.show()

// transform examines the input items against all the association rules and summarize the
// consequents as prediction
model.transform(dataset).show()

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

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

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

import org.apache.spark.ml.fpm.FPGrowth;
import org.apache.spark.ml.fpm.FPGrowthModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.*;

List<Row> data = Arrays.asList(
  RowFactory.create(Arrays.asList("1 2 5".split(" "))),
  RowFactory.create(Arrays.asList("1 2 3 5".split(" "))),
  RowFactory.create(Arrays.asList("1 2".split(" ")))
);
StructType schema = new StructType(new StructField[]{ new StructField(
  "items", new ArrayType(DataTypes.StringType, true), false, Metadata.empty())
});
Dataset<Row> itemsDF = spark.createDataFrame(data, schema);

FPGrowthModel model = new FPGrowth()
  .setItemsCol("items")
  .setMinSupport(0.5)
  .setMinConfidence(0.6)
  .fit(itemsDF);

// Display frequent itemsets.
model.freqItemsets().show();

// Display generated association rules.
model.associationRules().show();

// transform examines the input items against all the association rules and summarize the
// consequents as prediction
model.transform(itemsDF).show();

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

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

# Load training data

df <- selectExpr(createDataFrame(data.frame(rawItems = c(
  "1,2,5", "1,2,3,5", "1,2"
))), "split(rawItems, ',') AS items")

fpm <- spark.fpGrowth(df, itemsCol="items", minSupport=0.5, minConfidence=0.6)

# Extracting frequent itemsets

spark.freqItemsets(fpm)

# Extracting association rules

spark.associationRules(fpm)

# Predict uses association rules to and combines possible consequents

predict(fpm, df)

在 Spark 仓库中的 "examples/src/main/r/ml/fpm.R" 中找到完整的示例代码。

PrefixSpan

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

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

minSupport：被认为是频繁序列模式所需的最小支持度。
maxPatternLength：频繁序列模式的最大长度。任何超过此长度的频繁模式将不包含在结果中。
maxLocalProjDBSize：在开始本地迭代处理投影数据库之前，前缀投影数据库中允许的最大项数。应该根据执行程序的大小调整此参数。
sequenceCol：数据集中序列列的名称（默认值为 “sequence”），忽略此列中包含 null 值的行。

示例

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

from pyspark.ml.fpm import PrefixSpan

df = sc.parallelize([Row(sequence=[[1, 2], [3]]),
                     Row(sequence=[[1], [3, 2], [1, 2]]),
                     Row(sequence=[[1, 2], [5]]),
                     Row(sequence=[[6]])]).toDF()

prefixSpan = PrefixSpan(minSupport=0.5, maxPatternLength=5,
                        maxLocalProjDBSize=32000000)

# Find frequent sequential patterns.
prefixSpan.findFrequentSequentialPatterns(df).show()

在 Spark 仓库中的 "examples/src/main/python/ml/prefixspan_example.py" 中找到完整的示例代码。

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

import org.apache.spark.ml.fpm.PrefixSpan

val smallTestData = Seq(
  Seq(Seq(1, 2), Seq(3)),
  Seq(Seq(1), Seq(3, 2), Seq(1, 2)),
  Seq(Seq(1, 2), Seq(5)),
  Seq(Seq(6)))

val df = smallTestData.toDF("sequence")
val result = new PrefixSpan()
  .setMinSupport(0.5)
  .setMaxPatternLength(5)
  .setMaxLocalProjDBSize(32000000)
  .findFrequentSequentialPatterns(df)
  .show()

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

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

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

import org.apache.spark.ml.fpm.PrefixSpan;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.*;

List<Row> data = Arrays.asList(
  RowFactory.create(Arrays.asList(Arrays.asList(1, 2), Arrays.asList(3))),
  RowFactory.create(Arrays.asList(Arrays.asList(1), Arrays.asList(3, 2), Arrays.asList(1,2))),
  RowFactory.create(Arrays.asList(Arrays.asList(1, 2), Arrays.asList(5))),
  RowFactory.create(Arrays.asList(Arrays.asList(6)))
);
StructType schema = new StructType(new StructField[]{ new StructField(
  "sequence", new ArrayType(new ArrayType(DataTypes.IntegerType, true), true),
  false, Metadata.empty())
});
Dataset<Row> sequenceDF = spark.createDataFrame(data, schema);

PrefixSpan prefixSpan = new PrefixSpan().setMinSupport(0.5).setMaxPatternLength(5);

// Finding frequent sequential patterns
prefixSpan.findFrequentSequentialPatterns(sequenceDF).show();

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

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

# Load training data

df <- createDataFrame(list(list(list(list(1L, 2L), list(3L))),
                           list(list(list(1L), list(3L, 2L), list(1L, 2L))),
                           list(list(list(1L, 2L), list(5L))),
                           list(list(list(6L)))),
                      schema = c("sequence"))

# Finding frequent sequential patterns
frequency <- spark.findFrequentSequentialPatterns(df, minSupport = 0.5, maxPatternLength = 5L,
                                                  maxLocalProjDBSize = 32000000L)
showDF(frequency)

在 Spark 仓库中的 "examples/src/main/r/ml/prefixSpan.R" 中找到完整的示例代码。

MLlib：主要指南

MLlib：基于 RDD 的 API 指南

频繁模式挖掘

FP-Growth

PrefixSpan