Spark ML 14.聚类算法

1. K均值

1.1 算法介绍：

K 均值（K-means） 是一个常用的聚类算法来讲数据按预定的簇数进行剧集。k-means 算法的基本思想史： 以空间K个点为中心进行聚类， 对靠近他们的对象归类。 通过迭代的方法， 主次更新各聚类中心的值， 直至得到最好的聚类结果。

加谁要把样本分为C个类别， 算法描述如下：

1，适当选择c个类的初始中心

2，在第K次迭代中， 对任意一个样本求其到C个中心的距离， 将该样本归到距离最短的中心所在的类

3，利用均值等方法更新该类的中心值

4，对于所有的C个聚类中心， 如果利用2,3，的迭代方法更新后， 值保持不变， 则迭代结束，否则继续迭代。

MLlib 工具包包含并行的K-Means ++ 算法， 成为Kmeans||, Kmeans是一个Estimator， 它在基础模型之上产生一个KMeansModel

1.2 参数说明

参数名称	类型	说明
featuresCol	字符串	特征列名
predictionCol	字符串	预测结果列名
k	整数	聚类簇数
maxIter	整数	迭代次数（>=0）
seed	长整型	随机种子
tol	双精度	迭代算法的收敛性

1.3 调用示例

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

object KMeans extends App{
  val conf = new SparkConf().setAppName("KMeans")
  //设置master local[4] 指定本地模式开启模拟worker线程数
  conf.setMaster("local[4]")
  //创建sparkContext文件
  val sc = new SparkContext(conf)
  val spark = SparkSession.builder().getOrCreate()
  sc.setLogLevel("Error")

  val dataset = spark.read.format("libsvm").load("D:\\data\\sample_kmeans_data.txt")

  import org.apache.spark.ml.clustering.KMeans

  // Trains a k-means model.
  val kmeans = new KMeans().setK(2).setSeed(1L)
  val model = kmeans.fit(dataset)

  // 通过在平方误差平方和中计算评估聚类。
  val WSSSE = model.computeCost(dataset)
  println(s"Within Set Sum of Squared Errors = $WSSSE")
  //Within Set Sum of Squared Errors = 0.11999999999994547
  
  // Shows the result.
  println("Cluster Centers: ")
  model.clusterCenters.foreach(println)
  /**
  Cluster Centers:
  [0.1,0.1,0.1]
  [9.1,9.1,9.1]
    */
}

2. 文档主题生成模型(LDA)

2.1 算法介绍

LDA(Latent Dirichlet Allocation) 是一种文档主题生成模型， 也成为一个三层贝叶斯概率模型， 包含词，主题和文档三层结构。 所谓生成模型， 就是说， 我们认为一篇文章的每个词都是通过 “以一定概率选择了某个主题， 并从这个主题中以一定概率选择某个词语”这样一个过程得到。 文档到主题服从多项式分布， 主题到词服从多项式分布。

LDA是一种非监督机器学习技术， 可以用来识别大规模文档集（document collection）或语料库（corpus） 中潜藏的主题信息， 它采用了词袋（bag of words）的方法， 这种方法将每一篇文档视为一个词频向量， 从而将文本信息转化为了易于建模的数字信息。 但是词袋方法没有考虑词与词之间的顺序， 这简化了问题的复杂性， 同时也为模型的改进提供了契机。 每一篇文档代表了一些主题构成的一个概率分布， 而每一个主题又代表了很多单词所构成的概率分布。

2.2 参数说明

参数名称	类型	说明
featuresCol	字符串	特征列名
optimizer	字符串	估计LDA模型时使用的优化器
checkpointInterval	整数	设置检查点间隔（>=1），或不设置检查点（-1）
docConcentration	双精度数组	文档关于主题（”theta”）的先验分布集中参数（通常名为“alpha”）
k	整数	需推断的主题（簇）的数目
maxIter	整数	迭代次数（>=0）
seed	长整型	随机种子
subsamplingRate	双精度	仅对在线优化器（即optimizer=”online”）
topicConcentration	双精度	主题关于文字的先验分布集中参数（通常名为“beta”或”eta”）
topicDistributionCol	字符串型	每个文档的混合主题分布估计的输出列（文献中通常名为”theta”）

2.3 调用示例

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  * 文档主题生成模型(LDA)
  */
object b_LDA extends App{
  val conf = new SparkConf().setAppName("b_LDA")
  //设置master local[4] 指定本地模式开启模拟worker线程数
  conf.setMaster("local[4]")
  //创建sparkContext文件
  val sc = new SparkContext(conf)
  val spark = SparkSession.builder().getOrCreate()
  sc.setLogLevel("Error")

  val dataset = spark.read.format("libsvm").load("D:\\data\\sample_libsvm_data.txt")

  import org.apache.spark.ml.clustering.LDA

  // Trains a LDA model.
  val lda = new LDA().setK(10).setMaxIter(10)
  val model = lda.fit(dataset)

  val ll = model.logLikelihood(dataset)
  val lp = model.logPerplexity(dataset)
  println(s"The lower bound on the log likelihood of the entire corpus: $ll")
  //  The lower bound on the log likelihood of the entire corpus: -1.2923084327880664E7
  println(s"The upper bound bound on perplexity: $lp")
  //  The upper bound bound on perplexity: 5.308987518591441

  // 描述主题.
  val topics = model.describeTopics(3)
  println("The topics described by their top-weighted terms:")
  topics.show(3)
  /**
The topics described by their top-weighted terms:
+-----+---------------+--------------------+
|topic|    termIndices|         termWeights|
+-----+---------------+--------------------+
|    0|[597, 569, 598]|[0.01126906771778...|
|    1|[415, 398, 601]|[0.00950113504163...|
|    2|[261, 233, 260]|[0.01749002981877...|
+-----+---------------+--------------------+

    */
  // Shows the result.
  val transformed = model.transform(dataset)
  transformed.show(3)
  /**
+-----+--------------------+--------------------+
|label|            features|   topicDistribution|
+-----+--------------------+--------------------+
|  0.0|(692,[127,128,129...|[0.80140785875680...|
|  1.0|(692,[158,159,160...|[0.04601240753292...|
|  1.0|(692,[124,125,126...|[5.36847481605803...|
+-----+--------------------+--------------------+
    */
}

3. 二分K均值算法

3.1 算法介绍

二分K均值算法是一种层次聚类算法， 使用自顶向下的逼近， 所有的额观察值开始是一个簇，递归地向下层级分裂。分裂一句为选择能最大程度降低聚类代价函数（也就是误差平方和）的簇划分为两个簇。 以此进行下去， 知道簇的数目等于用户给定的数据K为止。 二分K均值常常比传统K均值算法有更快的计算速度， 但是产生的簇群与传统K均值算法旺旺也是不同的。

BisectingKMeans 是一个Estimator， 在基础模型上训练得到BisectingKMeansModel

3.2 参数说明

参数名称	类型	说明
featuresCol	字符串	特征列名
predictionCol	字符串	预测结果列名
k	整数	聚类簇数
maxIter	整数	迭代次数（>=0）
seed	长整型	随机种子

3.3 调用示例

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  * Created by admin on 2018/4/27.
  * 二分K均值算法
  */
object BisectingKMeans extends App{
  val conf = new SparkConf().setAppName("b_LDA")
  //设置master local[4] 指定本地模式开启模拟worker线程数
  conf.setMaster("local[4]")
  //创建sparkContext文件
  val sc = new SparkContext(conf)
  val spark = SparkSession.builder().getOrCreate()
  sc.setLogLevel("Error")

  val dataset = spark.read.format("libsvm").load("D:\\data\\sample_kmeans_data.txt")

  import org.apache.spark.ml.clustering.BisectingKMeans

  // Loads data.
  //val dataset = spark.read.format("libsvm").load("data/mllib/sample_kmeans_data.txt")

  // Trains a bisecting k-means model.
  val bkm = new BisectingKMeans().setK(2).setSeed(1)
  val model = bkm.fit(dataset)

  // Evaluate clustering.
  val cost = model.computeCost(dataset)
  println(s"Within Set Sum of Squared Errors = $cost")
	//Within Set Sum of Squared Errors = 0.11999999999994547
  // Shows the result.
  println("Cluster Centers: ")
  val centers = model.clusterCenters
  centers.foreach(println)
  /**
	Cluster Centers: 
	[0.1,0.1,0.1]
	[9.1,9.1,9.1]
    */

}

4. 高斯混合模型

4.1 算法介绍

混合高斯模型描述数据点以一定的概率服从K种高斯子分部的一种混合分部。 Spark.ml是用EM算法给出一组样本的极大似然模型。

4.2 参数介绍

参数名称	类型	说明
featuresCol	字符串	特征列名
probabilityCol	字符串	用以预测类别条件概率的列名
predictionCol	字符串	预测结果列名
k	整数	混合模型中独立的高斯数目
maxIter	整数	迭代次数（>=0）
seed	长整型	随机种子
tol	双精度	迭代算法的收敛性

4.3 调用示例

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  * 高斯混合模型
  */
object GaussianMixture extends App{
  val conf = new SparkConf().setAppName("d_GaussianMixture")
  //设置master local[4] 指定本地模式开启模拟worker线程数
  conf.setMaster("local[4]")
  //创建sparkContext文件
  val sc = new SparkContext(conf)
  val spark = SparkSession.builder().getOrCreate()
  sc.setLogLevel("Error")

  val dataset = spark.read.format("libsvm").load("D:\\data\\sample_kmeans_data.txt")

  import org.apache.spark.ml.clustering.GaussianMixture
  // Trains Gaussian Mixture Model
  val gmm = new GaussianMixture()
    .setK(2)
  val model = gmm.fit(dataset)
  
  // output parameters of mixture model model
  for (i <- 0 until model.getK) {
    println("weight=%f\nmu=%s\nsigma=\n%s\n" format
      (model.weights(i), model.gaussians(i).mean, model.gaussians(i).cov))
  }

  /**
 weight=0.500000
 mu=[0.10000000000001552,0.10000000000001552,0.10000000000001552]
 sigma=
 0.006666666666806455  0.006666666666806455  0.006666666666806455
 0.006666666666806455  0.006666666666806455  0.006666666666806455
 0.006666666666806455  0.006666666666806455  0.006666666666806455

 weight=0.500000
 mu=[9.099999999999985,9.099999999999985,9.099999999999985]
 sigma=
 0.006666666666783764  0.006666666666783764  0.006666666666783764
 0.006666666666783764  0.006666666666783764  0.006666666666783764
 0.006666666666783764  0.006666666666783764  0.006666666666783764
    */
}