聚类 - 基于 RDD 的 API

聚类是一种无监督学习问题，我们旨在根据某种相似性概念将实体子集相互分组。聚类通常用于探索性分析，或者作为分层监督学习管道的一部分（其中为每个集群训练不同的分类器或回归模型）。

spark.mllib 包支持以下模型

k 均值
高斯混合模型
幂迭代聚类 (PIC)
潜在狄利克雷分配 (LDA)
二分 k 均值
流式 k 均值

k 均值

k 均值是最常用的聚类算法之一，它将数据点聚类到预定义数量的集群中。 spark.mllib 实现包含 k 均值++ 方法的并行变体，称为 kmeans||。 spark.mllib 中的实现具有以下参数

k 是所需集群的数量。请注意，返回的集群数量可能少于 k，例如，如果要聚类的不同点少于 k。
maxIterations 是要运行的最大迭代次数。
initializationMode 指定随机初始化或通过 k 均值|| 初始化。
runs 此参数从 Spark 2.0.0 开始无效。
initializationSteps 确定 k 均值|| 算法中的步骤数。
epsilon 确定我们认为 k 均值已收敛的距离阈值。
initialModel 是一组可选的集群中心，用于初始化。如果提供此参数，则只执行一次运行。

示例

以下示例可以在 PySpark shell 中测试。

在以下示例中，在加载和解析数据后，我们使用 KMeans 对象将数据聚类到两个集群中。所需集群的数量传递给算法。然后我们计算组内平方和误差 (WSSSE)。您可以通过增加 k 来减少此误差度量。实际上，最佳 k 通常是在 WSSSE 图中出现“肘部”的位置。

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

from numpy import array
from math import sqrt

from pyspark.mllib.clustering import KMeans, KMeansModel

# Load and parse the data
data = sc.textFile("data/mllib/kmeans_data.txt")
parsedData = data.map(lambda line: array([float(x) for x in line.split(' ')]))

# Build the model (cluster the data)
clusters = KMeans.train(parsedData, 2, maxIterations=10, initializationMode="random")

# Evaluate clustering by computing Within Set Sum of Squared Errors
def error(point):
    center = clusters.centers[clusters.predict(point)]
    return sqrt(sum([x**2 for x in (point - center)]))

WSSSE = parsedData.map(lambda point: error(point)).reduce(lambda x, y: x + y)
print("Within Set Sum of Squared Error = " + str(WSSSE))

# Save and load model
clusters.save(sc, "target/org/apache/spark/PythonKMeansExample/KMeansModel")
sameModel = KMeansModel.load(sc, "target/org/apache/spark/PythonKMeansExample/KMeansModel")

在 Spark 存储库的“examples/src/main/python/mllib/k_means_example.py”中找到完整的示例代码。

以下代码段可以在 spark-shell 中执行。

在以下示例中，在加载和解析数据后，我们使用 KMeans 对象将数据聚类到两个集群中。所需集群的数量传递给算法。然后我们计算组内平方和误差 (WSSSE)。您可以通过增加 k 来减少此误差度量。实际上，最佳 k 通常是在 WSSSE 图中出现“肘部”的位置。

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

import org.apache.spark.mllib.clustering.{KMeans, KMeansModel}
import org.apache.spark.mllib.linalg.Vectors

// Load and parse the data
val data = sc.textFile("data/mllib/kmeans_data.txt")
val parsedData = data.map(s => Vectors.dense(s.split(' ').map(_.toDouble))).cache()

// Cluster the data into two classes using KMeans
val numClusters = 2
val numIterations = 20
val clusters = KMeans.train(parsedData, numClusters, numIterations)

// Evaluate clustering by computing Within Set Sum of Squared Errors
val WSSSE = clusters.computeCost(parsedData)
println(s"Within Set Sum of Squared Errors = $WSSSE")

// Save and load model
clusters.save(sc, "target/org/apache/spark/KMeansExample/KMeansModel")
val sameModel = KMeansModel.load(sc, "target/org/apache/spark/KMeansExample/KMeansModel")

在 Spark 存储库的“examples/src/main/scala/org/apache/spark/examples/mllib/KMeansExample.scala”中找到完整的示例代码。

所有 MLlib 的方法都使用 Java 友好类型，因此您可以在 Java 中以与在 Scala 中相同的方式导入和调用它们。唯一的警告是，这些方法接受 Scala RDD 对象，而 Spark Java API 使用单独的 JavaRDD 类。您可以通过在 JavaRDD 对象上调用 .rdd() 来将 Java RDD 转换为 Scala RDD。下面给出了与 Scala 中提供的示例等效的独立应用程序示例

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

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.clustering.KMeans;
import org.apache.spark.mllib.clustering.KMeansModel;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;

// Load and parse data
String path = "data/mllib/kmeans_data.txt";
JavaRDD<String> data = jsc.textFile(path);
JavaRDD<Vector> parsedData = data.map(s -> {
  String[] sarray = s.split(" ");
  double[] values = new double[sarray.length];
  for (int i = 0; i < sarray.length; i++) {
    values[i] = Double.parseDouble(sarray[i]);
  }
  return Vectors.dense(values);
});
parsedData.cache();

// Cluster the data into two classes using KMeans
int numClusters = 2;
int numIterations = 20;
KMeansModel clusters = KMeans.train(parsedData.rdd(), numClusters, numIterations);

System.out.println("Cluster centers:");
for (Vector center: clusters.clusterCenters()) {
  System.out.println(" " + center);
}
double cost = clusters.computeCost(parsedData.rdd());
System.out.println("Cost: " + cost);

// Evaluate clustering by computing Within Set Sum of Squared Errors
double WSSSE = clusters.computeCost(parsedData.rdd());
System.out.println("Within Set Sum of Squared Errors = " + WSSSE);

// Save and load model
clusters.save(jsc.sc(), "target/org/apache/spark/JavaKMeansExample/KMeansModel");
KMeansModel sameModel = KMeansModel.load(jsc.sc(),
  "target/org/apache/spark/JavaKMeansExample/KMeansModel");

在 Spark 存储库的“examples/src/main/java/org/apache/spark/examples/mllib/JavaKMeansExample.java”中找到完整的示例代码。

高斯混合模型

高斯混合模型表示一个复合分布，其中点从 k 个高斯子分布中的一个中抽取，每个子分布都有自己的概率。 spark.mllib 实现使用期望最大化算法来推断给定一组样本的最大似然模型。该实现具有以下参数

k 是所需集群的数量。
convergenceTol 是我们认为收敛已实现的对数似然度的最大变化。
maxIterations 是在达到收敛之前要执行的最大迭代次数。
initialModel 是一个可选的起点，从该起点开始 EM 算法。如果省略此参数，则将从数据中构建一个随机起点。

示例

在以下示例中，在加载和解析数据后，我们使用 GaussianMixture 对象将数据聚类到两个集群中。所需集群的数量传递给算法。然后我们输出混合模型的参数。

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

from numpy import array

from pyspark.mllib.clustering import GaussianMixture, GaussianMixtureModel

# Load and parse the data
data = sc.textFile("data/mllib/gmm_data.txt")
parsedData = data.map(lambda line: array([float(x) for x in line.strip().split(' ')]))

# Build the model (cluster the data)
gmm = GaussianMixture.train(parsedData, 2)

# Save and load model
gmm.save(sc, "target/org/apache/spark/PythonGaussianMixtureExample/GaussianMixtureModel")
sameModel = GaussianMixtureModel\
    .load(sc, "target/org/apache/spark/PythonGaussianMixtureExample/GaussianMixtureModel")

# output parameters of model
for i in range(2):
    print("weight = ", gmm.weights[i], "mu = ", gmm.gaussians[i].mu,
          "sigma = ", gmm.gaussians[i].sigma.toArray())

在 Spark 存储库的“examples/src/main/python/mllib/gaussian_mixture_example.py”中找到完整的示例代码。

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

import org.apache.spark.mllib.clustering.{GaussianMixture, GaussianMixtureModel}
import org.apache.spark.mllib.linalg.Vectors

// Load and parse the data
val data = sc.textFile("data/mllib/gmm_data.txt")
val parsedData = data.map(s => Vectors.dense(s.trim.split(' ').map(_.toDouble))).cache()

// Cluster the data into two classes using GaussianMixture
val gmm = new GaussianMixture().setK(2).run(parsedData)

// Save and load model
gmm.save(sc, "target/org/apache/spark/GaussianMixtureExample/GaussianMixtureModel")
val sameModel = GaussianMixtureModel.load(sc,
  "target/org/apache/spark/GaussianMixtureExample/GaussianMixtureModel")

// output parameters of max-likelihood model
for (i <- 0 until gmm.k) {
  println("weight=%f\nmu=%s\nsigma=\n%s\n" format
    (gmm.weights(i), gmm.gaussians(i).mu, gmm.gaussians(i).sigma))
}

在 Spark 存储库的“examples/src/main/scala/org/apache/spark/examples/mllib/GaussianMixtureExample.scala”中找到完整的示例代码。

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

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.clustering.GaussianMixture;
import org.apache.spark.mllib.clustering.GaussianMixtureModel;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;

// Load and parse data
String path = "data/mllib/gmm_data.txt";
JavaRDD<String> data = jsc.textFile(path);
JavaRDD<Vector> parsedData = data.map(s -> {
  String[] sarray = s.trim().split(" ");
  double[] values = new double[sarray.length];
  for (int i = 0; i < sarray.length; i++) {
    values[i] = Double.parseDouble(sarray[i]);
  }
  return Vectors.dense(values);
});
parsedData.cache();

// Cluster the data into two classes using GaussianMixture
GaussianMixtureModel gmm = new GaussianMixture().setK(2).run(parsedData.rdd());

// Save and load GaussianMixtureModel
gmm.save(jsc.sc(), "target/org/apache/spark/JavaGaussianMixtureExample/GaussianMixtureModel");
GaussianMixtureModel sameModel = GaussianMixtureModel.load(jsc.sc(),
  "target/org.apache.spark.JavaGaussianMixtureExample/GaussianMixtureModel");

// Output the parameters of the mixture model
for (int j = 0; j < gmm.k(); j++) {
  System.out.printf("weight=%f\nmu=%s\nsigma=\n%s\n",
    gmm.weights()[j], gmm.gaussians()[j].mu(), gmm.gaussians()[j].sigma());
}

在 Spark 存储库的“examples/src/main/java/org/apache/spark/examples/mllib/JavaGaussianMixtureExample.java”中找到完整的示例代码。

幂迭代聚类 (PIC)

幂迭代聚类 (PIC) 是一种可扩展且高效的算法，用于在给定成对相似性作为边属性的情况下对图的顶点进行聚类，如 Lin and Cohen, Power Iteration Clustering 中所述。它通过幂迭代计算图的归一化亲和矩阵的伪特征向量，并使用它来对顶点进行聚类。 spark.mllib 包含使用 GraphX 作为其后端的 PIC 实现。它接受 (srcId, dstId, similarity) 元组的 RDD 并输出具有聚类分配的模型。相似性必须是非负的。PIC 假设相似性度量是对称的。无论排序如何，对 (srcId, dstId) 应该在输入数据中最多出现一次。如果输入中缺少一对，则它们的相似性被视为零。 spark.mllib 的 PIC 实现采用以下（超）参数

k：集群数量
maxIterations：最大幂迭代次数
initializationMode：初始化模型。这可以是“random”，这是默认值，使用随机向量作为顶点属性，或者“degree”，使用归一化的总相似性。

示例

在下面，我们展示了代码片段来演示如何在 spark.mllib 中使用 PIC。

PowerIterationClustering 实现 PIC 算法。它接受一个表示亲和矩阵的 (srcId: Long, dstId: Long, similarity: Double) 元组的 RDD。调用 PowerIterationClustering.run 返回一个 PowerIterationClusteringModel，其中包含计算出的聚类分配。

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

from pyspark.mllib.clustering import PowerIterationClustering, PowerIterationClusteringModel

# Load and parse the data
data = sc.textFile("data/mllib/pic_data.txt")
similarities = data.map(lambda line: tuple([float(x) for x in line.split(' ')]))

# Cluster the data into two classes using PowerIterationClustering
model = PowerIterationClustering.train(similarities, 2, 10)

model.assignments().foreach(lambda x: print(str(x.id) + " -> " + str(x.cluster)))

# Save and load model
model.save(sc, "target/org/apache/spark/PythonPowerIterationClusteringExample/PICModel")
sameModel = PowerIterationClusteringModel\
    .load(sc, "target/org/apache/spark/PythonPowerIterationClusteringExample/PICModel")

在 Spark 存储库中的“examples/src/main/python/mllib/power_iteration_clustering_example.py”中找到完整的示例代码。

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

import org.apache.spark.mllib.clustering.PowerIterationClustering

val circlesRdd = generateCirclesRdd(sc, params.k, params.numPoints)
val model = new PowerIterationClustering()
  .setK(params.k)
  .setMaxIterations(params.maxIterations)
  .setInitializationMode("degree")
  .run(circlesRdd)

val clusters = model.assignments.collect().groupBy(_.cluster).mapValues(_.map(_.id))
val assignments = clusters.toList.sortBy { case (k, v) => v.length }
val assignmentsStr = assignments
  .map { case (k, v) =>
    s"$k -> ${v.sorted.mkString("[", ",", "]")}"
  }.mkString(", ")
val sizesStr = assignments.map {
  _._2.length
}.sorted.mkString("(", ",", ")")
println(s"Cluster assignments: $assignmentsStr\ncluster sizes: $sizesStr")

在 Spark 存储库中的“examples/src/main/scala/org/apache/spark/examples/mllib/PowerIterationClusteringExample.scala”中找到完整的示例代码。

PowerIterationClustering 实现 PIC 算法。它接受一个表示亲和矩阵的 (srcId: Long, dstId: Long, similarity: Double) 元组的 JavaRDD。调用 PowerIterationClustering.run 返回一个 PowerIterationClusteringModel，其中包含计算出的聚类分配。

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

import org.apache.spark.mllib.clustering.PowerIterationClustering;
import org.apache.spark.mllib.clustering.PowerIterationClusteringModel;

JavaRDD<Tuple3<Long, Long, Double>> similarities = sc.parallelize(Arrays.asList(
  new Tuple3<>(0L, 1L, 0.9),
  new Tuple3<>(1L, 2L, 0.9),
  new Tuple3<>(2L, 3L, 0.9),
  new Tuple3<>(3L, 4L, 0.1),
  new Tuple3<>(4L, 5L, 0.9)));

PowerIterationClustering pic = new PowerIterationClustering()
  .setK(2)
  .setMaxIterations(10);
PowerIterationClusteringModel model = pic.run(similarities);

for (PowerIterationClustering.Assignment a: model.assignments().toJavaRDD().collect()) {
  System.out.println(a.id() + " -> " + a.cluster());
}

在 Spark 存储库中的“examples/src/main/java/org/apache/spark/examples/mllib/JavaPowerIterationClusteringExample.java”中找到完整的示例代码。

潜在狄利克雷分配 (LDA)

潜在狄利克雷分配 (LDA) 是一种主题模型，它从文本文档集合中推断主题。LDA 可以被认为是一种聚类算法，如下所示

主题对应于聚类中心，文档对应于数据集中示例（行）。
主题和文档都存在于特征空间中，其中特征向量是词频向量（词袋）。
LDA 不使用传统的距离来估计聚类，而是使用基于文本文档生成统计模型的函数。

LDA 通过 setOptimizer 函数支持不同的推理算法。 EMLDAOptimizer 使用期望最大化在似然函数上学习聚类，并产生全面结果，而 OnlineLDAOptimizer 使用迭代小批量采样进行在线变分推理，通常对内存友好。

LDA 接受词频向量形式的文档集合以及以下参数（使用构建器模式设置）

k：主题数量（即聚类中心）
optimizer：用于学习 LDA 模型的优化器，可以是 EMLDAOptimizer 或 OnlineLDAOptimizer
docConcentration：文档对主题分布的先验的狄利克雷参数。较大的值鼓励更平滑的推断分布。
topicConcentration：主题对词语（词语）分布的先验的狄利克雷参数。较大的值鼓励更平滑的推断分布。
maxIterations：迭代次数限制。
checkpointInterval：如果使用检查点（在 Spark 配置中设置），此参数指定创建检查点的频率。如果 maxIterations 很大，使用检查点可以帮助减少磁盘上的混洗文件大小，并有助于故障恢复。

所有 spark.mllib 的 LDA 模型都支持

describeTopics：将主题作为最重要的词语和词语权重的数组返回
topicsMatrix：返回一个 vocabSize 乘以 k 的矩阵，其中每一列都是一个主题

注意：LDA 仍然是一个处于积极开发中的实验性功能。因此，某些功能仅在优化器生成的两个优化器/模型之一中可用。目前，分布式模型可以转换为本地模型，反之则不行。

以下讨论将分别描述每个优化器/模型对。

期望最大化

在 EMLDAOptimizer 和 DistributedLDAModel 中实现。

对于提供给 LDA 的参数

docConcentration：仅支持对称先验，因此提供的 k 维向量中的所有值必须相同。所有值也必须大于 1.0。提供 Vector(-1) 会导致默认行为（具有值 $(50 / k) + 1$ 的均匀 k 维向量）
topicConcentration：仅支持对称先验。值必须大于 1.0。提供 -1 会导致默认为值 0.1 + 1。
maxIterations：EM 迭代的最大次数。

注意：进行足够的迭代很重要。在早期迭代中，EM 通常具有无用的主题，但这些主题在更多迭代后会显着改善。使用至少 20 次，可能 50-100 次迭代通常是合理的，具体取决于您的数据集。

EMLDAOptimizer 生成一个 DistributedLDAModel，它不仅存储推断的主题，还存储完整的训练语料库以及训练语料库中每个文档的主题分布。 DistributedLDAModel 支持

topTopicsPerDocument：训练语料库中每个文档的顶级主题及其权重
topDocumentsPerTopic：每个主题的顶级文档以及主题在文档中的对应权重。
logPrior：给定超参数 docConcentration 和 topicConcentration，估计的主题和文档-主题分布的对数概率
logLikelihood：给定推断的主题和文档-主题分布，训练语料库的对数似然

在线变分贝叶斯

在 OnlineLDAOptimizer 和 LocalLDAModel 中实现。

对于提供给 LDA 的参数

docConcentration：可以通过传递一个向量来使用非对称先验，该向量的值等于 k 个维度中的每个维度的狄利克雷参数。值应大于等于 0。提供 Vector(-1) 会导致默认行为（具有值 $(1.0 / k)$ 的均匀 k 维向量）
topicConcentration：仅支持对称先验。值必须大于等于 0。提供 -1 会导致默认为值 $(1.0 / k)$。
maxIterations：要提交的小批次的最大数量。

此外，OnlineLDAOptimizer 接受以下参数

miniBatchFraction：每次迭代采样并使用的语料库的比例
optimizeDocConcentration：如果设置为 true，则在每个小批量之后执行超参数 docConcentration（也称为 alpha）的最大似然估计，并在返回的 LocalLDAModel 中设置优化的 docConcentration
tau0 和 kappa：用于学习率衰减，学习率衰减由 $(\tau_0 + iter)^{-\kappa}$ 计算，其中 $iter$ 是当前迭代次数。

OnlineLDAOptimizer 生成一个 LocalLDAModel，它只存储推断的主题。 LocalLDAModel 支持

logLikelihood(documents)：计算给定推断的主题的提供的 documents 的下界。
logPerplexity(documents)：计算给定推断的主题的提供的 documents 的困惑度的上界。

示例

在以下示例中，我们加载表示文档语料库的词频向量。然后，我们使用 LDA 从文档中推断出三个主题。所需聚类数量传递给算法。然后，我们输出主题，表示为词语上的概率分布。

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

from pyspark.mllib.clustering import LDA, LDAModel
from pyspark.mllib.linalg import Vectors

# Load and parse the data
data = sc.textFile("data/mllib/sample_lda_data.txt")
parsedData = data.map(lambda line: Vectors.dense([float(x) for x in line.strip().split(' ')]))
# Index documents with unique IDs
corpus = parsedData.zipWithIndex().map(lambda x: [x[1], x[0]]).cache()

# Cluster the documents into three topics using LDA
ldaModel = LDA.train(corpus, k=3)

# Output topics. Each is a distribution over words (matching word count vectors)
print("Learned topics (as distributions over vocab of " + str(ldaModel.vocabSize())
      + " words):")
topics = ldaModel.topicsMatrix()
for topic in range(3):
    print("Topic " + str(topic) + ":")
    for word in range(0, ldaModel.vocabSize()):
        print(" " + str(topics[word][topic]))

# Save and load model
ldaModel.save(sc, "target/org/apache/spark/PythonLatentDirichletAllocationExample/LDAModel")
sameModel = LDAModel\
    .load(sc, "target/org/apache/spark/PythonLatentDirichletAllocationExample/LDAModel")

在 Spark 存储库中的“examples/src/main/python/mllib/latent_dirichlet_allocation_example.py”中找到完整的示例代码。

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

import org.apache.spark.mllib.clustering.{DistributedLDAModel, LDA}
import org.apache.spark.mllib.linalg.Vectors

// Load and parse the data
val data = sc.textFile("data/mllib/sample_lda_data.txt")
val parsedData = data.map(s => Vectors.dense(s.trim.split(' ').map(_.toDouble)))
// Index documents with unique IDs
val corpus = parsedData.zipWithIndex.map(_.swap).cache()

// Cluster the documents into three topics using LDA
val ldaModel = new LDA().setK(3).run(corpus)

// Output topics. Each is a distribution over words (matching word count vectors)
println(s"Learned topics (as distributions over vocab of ${ldaModel.vocabSize} words):")
val topics = ldaModel.topicsMatrix
for (topic <- Range(0, 3)) {
  print(s"Topic $topic :")
  for (word <- Range(0, ldaModel.vocabSize)) {
    print(s"${topics(word, topic)}")
  }
  println()
}

// Save and load model.
ldaModel.save(sc, "target/org/apache/spark/LatentDirichletAllocationExample/LDAModel")
val sameModel = DistributedLDAModel.load(sc,
  "target/org/apache/spark/LatentDirichletAllocationExample/LDAModel")

在 Spark 存储库中的“examples/src/main/scala/org/apache/spark/examples/mllib/LatentDirichletAllocationExample.scala”中找到完整的示例代码。

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

import scala.Tuple2;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.clustering.DistributedLDAModel;
import org.apache.spark.mllib.clustering.LDA;
import org.apache.spark.mllib.clustering.LDAModel;
import org.apache.spark.mllib.linalg.Matrix;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;

// Load and parse the data
String path = "data/mllib/sample_lda_data.txt";
JavaRDD<String> data = jsc.textFile(path);
JavaRDD<Vector> parsedData = data.map(s -> {
  String[] sarray = s.trim().split(" ");
  double[] values = new double[sarray.length];
  for (int i = 0; i < sarray.length; i++) {
    values[i] = Double.parseDouble(sarray[i]);
  }
  return Vectors.dense(values);
});
// Index documents with unique IDs
JavaPairRDD<Long, Vector> corpus =
  JavaPairRDD.fromJavaRDD(parsedData.zipWithIndex().map(Tuple2::swap));
corpus.cache();

// Cluster the documents into three topics using LDA
LDAModel ldaModel = new LDA().setK(3).run(corpus);

// Output topics. Each is a distribution over words (matching word count vectors)
System.out.println("Learned topics (as distributions over vocab of " + ldaModel.vocabSize()
  + " words):");
Matrix topics = ldaModel.topicsMatrix();
for (int topic = 0; topic < 3; topic++) {
  System.out.print("Topic " + topic + ":");
  for (int word = 0; word < ldaModel.vocabSize(); word++) {
    System.out.print(" " + topics.apply(word, topic));
  }
  System.out.println();
}

ldaModel.save(jsc.sc(),
  "target/org/apache/spark/JavaLatentDirichletAllocationExample/LDAModel");
DistributedLDAModel sameModel = DistributedLDAModel.load(jsc.sc(),
  "target/org/apache/spark/JavaLatentDirichletAllocationExample/LDAModel");

在 Spark 存储库中的“examples/src/main/java/org/apache/spark/examples/mllib/JavaLatentDirichletAllocationExample.java”中找到完整的示例代码。

二分 k 均值

二分 k 均值通常比常规 k 均值快得多，但它通常会产生不同的聚类。

二分 k 均值是一种层次聚类。层次聚类是最常用的聚类分析方法之一，它试图构建一个聚类层次结构。层次聚类策略通常分为两种类型

凝聚式：这是一种“自下而上”的方法：每个观测值都从自己的集群开始，随着层次结构的上升，成对的集群被合并。
分裂式：这是一种“自上而下”的方法：所有观测值都从一个集群开始，随着层次结构的下降，递归地进行拆分。

二分 k 均值算法是一种分裂式算法。MLlib 中的实现具有以下参数

k：所需的叶集群数量（默认值：4）。如果不存在可分割的叶集群，实际数量可能更小。
maxIterations：拆分集群的最大 k 均值迭代次数（默认值：20）
minDivisibleClusterSize：可分割集群的最小点数（如果 >= 1.0）或最小比例（如果 < 1.0）（默认值：1）
seed：随机种子（默认值：类名的哈希值）

示例

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

from numpy import array

from pyspark.mllib.clustering import BisectingKMeans

# Load and parse the data
data = sc.textFile("data/mllib/kmeans_data.txt")
parsedData = data.map(lambda line: array([float(x) for x in line.split(' ')]))

# Build the model (cluster the data)
model = BisectingKMeans.train(parsedData, 2, maxIterations=5)

# Evaluate clustering
cost = model.computeCost(parsedData)
print("Bisecting K-means Cost = " + str(cost))

在 Spark 存储库中的“examples/src/main/python/mllib/bisecting_k_means_example.py”中找到完整的示例代码。

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

import org.apache.spark.mllib.clustering.BisectingKMeans
import org.apache.spark.mllib.linalg.{Vector, Vectors}

// Loads and parses data
def parse(line: String): Vector = Vectors.dense(line.split(" ").map(_.toDouble))
val data = sc.textFile("data/mllib/kmeans_data.txt").map(parse).cache()

// Clustering the data into 6 clusters by BisectingKMeans.
val bkm = new BisectingKMeans().setK(6)
val model = bkm.run(data)

// Show the compute cost and the cluster centers
println(s"Compute Cost: ${model.computeCost(data)}")
model.clusterCenters.zipWithIndex.foreach { case (center, idx) =>
  println(s"Cluster Center ${idx}: ${center}")
}

在 Spark 存储库中的“examples/src/main/scala/org/apache/spark/examples/mllib/BisectingKMeansExample.scala”中找到完整的示例代码。

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

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

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.clustering.BisectingKMeans;
import org.apache.spark.mllib.clustering.BisectingKMeansModel;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;

List<Vector> localData = Arrays.asList(
  Vectors.dense(0.1, 0.1),   Vectors.dense(0.3, 0.3),
  Vectors.dense(10.1, 10.1), Vectors.dense(10.3, 10.3),
  Vectors.dense(20.1, 20.1), Vectors.dense(20.3, 20.3),
  Vectors.dense(30.1, 30.1), Vectors.dense(30.3, 30.3)
);
JavaRDD<Vector> data = sc.parallelize(localData, 2);

BisectingKMeans bkm = new BisectingKMeans()
  .setK(4);
BisectingKMeansModel model = bkm.run(data);

System.out.println("Compute Cost: " + model.computeCost(data));

Vector[] clusterCenters = model.clusterCenters();
for (int i = 0; i < clusterCenters.length; i++) {
  Vector clusterCenter = clusterCenters[i];
  System.out.println("Cluster Center " + i + ": " + clusterCenter);
}

在 Spark 存储库中的“examples/src/main/java/org/apache/spark/examples/mllib/JavaBisectingKMeansExample.java”中找到完整的示例代码。

流式 k 均值

当数据以流的形式到达时，我们可能希望动态地估计集群，并在新数据到达时更新它们。 spark.mllib 提供对流式 k 均值聚类的支持，并提供参数来控制估计值的衰减（或“遗忘”）。该算法使用小批量 k 均值更新规则的泛化。对于每批数据，我们将所有点分配到其最近的集群，计算新的集群中心，然后使用以下公式更新每个集群

\begin{equation} c_{t+1} = \frac{c_tn_t\alpha + x_tm_t}{n_t\alpha+m_t} \end{equation} \begin{equation} n_{t+1} = n_t + m_t \end{equation}

其中 $c_t$ 是集群的先前中心， $n_t$ 是迄今为止分配到集群的点数， $x_t$ 是当前批次的新的集群中心， $m_t$ 是当前批次中添加到集群的点数。衰减因子 $\alpha$ 可用于忽略过去：当 $\alpha$=1 时，将使用从一开始的所有数据；当 $\alpha$=0 时，将仅使用最新数据。这类似于指数加权移动平均。

可以使用 halfLife 参数指定衰减，该参数确定正确的衰减因子 a，使得对于在时间 t 处获取的数据，其在时间 t + halfLife 处的贡献将下降到 0.5。时间单位可以指定为 batches 或 points，并且更新规则将相应地调整。

示例

此示例演示了如何在流式数据上估计集群。

有关 API 的更多详细信息，请参阅 StreamingKMeans Python 文档。有关 StreamingContext 的详细信息，请参阅 Spark Streaming 编程指南。

from pyspark.mllib.linalg import Vectors
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.clustering import StreamingKMeans

# we make an input stream of vectors for training,
# as well as a stream of vectors for testing
def parse(lp):
    label = float(lp[lp.find('(') + 1: lp.find(')')])
    vec = Vectors.dense(lp[lp.find('[') + 1: lp.find(']')].split(','))

    return LabeledPoint(label, vec)

trainingData = sc.textFile("data/mllib/kmeans_data.txt")\
    .map(lambda line: Vectors.dense([float(x) for x in line.strip().split(' ')]))

testingData = sc.textFile("data/mllib/streaming_kmeans_data_test.txt").map(parse)

trainingQueue = [trainingData]
testingQueue = [testingData]

trainingStream = ssc.queueStream(trainingQueue)
testingStream = ssc.queueStream(testingQueue)

# We create a model with random clusters and specify the number of clusters to find
model = StreamingKMeans(k=2, decayFactor=1.0).setRandomCenters(3, 1.0, 0)

# Now register the streams for training and testing and start the job,
# printing the predicted cluster assignments on new data points as they arrive.
model.trainOn(trainingStream)

result = model.predictOnValues(testingStream.map(lambda lp: (lp.label, lp.features)))
result.pprint()

ssc.start()
ssc.stop(stopSparkContext=True, stopGraceFully=True)

在 Spark 存储库中的“examples/src/main/python/mllib/streaming_k_means_example.py”中找到完整的示例代码。

有关 API 的详细信息，请参阅 StreamingKMeans Scala 文档。有关 StreamingContext 的详细信息，请参阅 Spark Streaming 编程指南。

import org.apache.spark.mllib.clustering.StreamingKMeans
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.streaming.{Seconds, StreamingContext}

val conf = new SparkConf().setAppName("StreamingKMeansExample")
val ssc = new StreamingContext(conf, Seconds(args(2).toLong))

val trainingData = ssc.textFileStream(args(0)).map(Vectors.parse)
val testData = ssc.textFileStream(args(1)).map(LabeledPoint.parse)

val model = new StreamingKMeans()
  .setK(args(3).toInt)
  .setDecayFactor(1.0)
  .setRandomCenters(args(4).toInt, 0.0)

model.trainOn(trainingData)
model.predictOnValues(testData.map(lp => (lp.label, lp.features))).print()

ssc.start()
ssc.awaitTermination()

在 Spark 存储库中的“examples/src/main/scala/org/apache/spark/examples/mllib/StreamingKMeansExample.scala”中找到完整的示例代码。

随着您添加包含数据的新的文本文件，集群中心将更新。每个训练点应格式化为 [x1, x2, x3]，每个测试数据点应格式化为 (y, [x1, x2, x3])，其中 y 是某个有用的标签或标识符（例如，真实的类别分配）。只要将文本文件放置在 /training/data/dir 中，模型就会更新。只要将文本文件放置在 /testing/data/dir 中，您就会看到预测。随着新数据的出现，集群中心将发生变化！

MLlib：主要指南

MLlib：基于 RDD 的 API 指南