如何在 Spark ML 中实现 Kmeans 评估器

Posted 2023-03-12

【中文标题】如何在 Spark ML 中实现 Kmeans 评估器

【英文标题】：How to implement Kmeans evaluator in Spark ML

【发布时间】：2018-05-17 17:51:03

【问题描述】：

我想根据 'k' 参数根据最低 k-means 分数选择 k-means 模型。

我可以手动找到'k'参数的最佳值，写类似

  def clusteringScore0(data: DataFrame, k: Int): Double = 
    val assembler = new VectorAssembler().
      setInputCols(data.columns.filter(_ != "label")).
      setOutputCol("featureVector")
    val kmeans = new KMeans().
      setSeed(Random.nextLong()).
      setK(k).
      setPredictionCol("cluster").
      setFeaturesCol("featureVector")
    val pipeline = new Pipeline().setStages(Array(assembler, kmeans))
    val kmeansModel = pipeline.fit(data).stages.last.asInstanceOf[KMeansModel]
    kmeansModel.computeCost(assembler.transform(data)) / data.count()   

  (20 to 100 by 20).map(k => (k, clusteringScore0(numericOnly, k))).
    foreach(println)

---

我应该使用 CrossValitor API 吗？

类似这样的：

val paramGrid = new ParamGridBuilder().addGrid(kmeansModel.k, 20 to 100 by 20).build()
val cv = new CrossValidator().setEstimator(pipeline).setEvaluator(new KMeansEvaluator()).setEstimatorParamMaps(paramGrid).setNumFolds(3)

有用于回归和分类的评估器，但没有用于聚类的评估器。

所以我应该实现 Evaluator 接口。我坚持使用evaluate 方法。

class KMeansEvaluator extends Evaluator 
  override def copy(extra: ParamMap): Evaluator = defaultCopy(extra)

  override def evaluate(data: Dataset[_]): Double = ??? // should I somehow adapt code from KMeansModel.computeCost()?
  override val uid = Identifiable.randomUID("cost_evaluator")

【参考方案1】：

您好 ClusteringEvaluator 可从 Spark 2.3.0 获得。您可以通过将 ClusteringEvaluator 对象包含到您的 for 循环中来找到最佳 k 值。您还可以在Scikit-learn page 中找到更多关于轮廓分析的详细信息。总之，分数应该在[-1,1]之间，分数越大越好。我为您的代码修改了下面的 for 循环。

import org.apache.spark.ml.evaluation.ClusteringEvaluator
val evaluator = new ClusteringEvaluator()
        .setFeaturesCol("featureVector")
        .setPredictionCol("cluster")
        .setMetricName("silhouette")

for(k <- 20 to 100 by 20)
    clusteringScore0(numericOnly,k)

    val transformedDF = kmeansModel.transform(numericOnly)

    val score = evaluator.evaluate(transformedDF)

    println(k,score,kmeansModel.computeCost(transformedDF))