Spark MLlib基于K-Means实现数据聚类

浅浅的花香味﹌ 2022-11-29 13:03 113阅读 0赞

### Spark MLlib基于K-Means实现数据聚类 ###

*  K-Means
 *   *  K-Means简介
     *  算法实现原则
 *  Spark实现K-Means
 *   *  Spark机器学习库
     *  算法步骤
     *  完整代码（基于spark.ml）

# K-Means #

## K-Means简介 ##

k均值聚类算法（k-means clustering algorithm）是一种迭代求解的聚类分析算法，其步骤是，预将数据分为K组，则随机选取K个对象作为初始的聚类中心，然后计算每个对象与各个种子聚类中心之间的距离，把每个对象分配给距离它最近的聚类中心。聚类中心以及分配给它们的对象就代表一个聚类。每分配一个样本，聚类的聚类中心会根据聚类中现有的对象被重新计算。这个过程将不断重复直到满足某个终止条件。终止条件可以是没有（或最小数目）对象被重新分配给不同的聚类，没有（或最小数目）聚类中心再发生变化，误差平方和局部最小。

## 算法实现原则 ##

先随机选取K个对象作为初始的聚类中心。然后计算每个对象与各个种子聚类中心之间的距离，把每个对象分配给距离它最近的聚类中心。聚类中心以及分配给它们的对象就代表一个聚类。一旦全部对象都被分配了，每个聚类的聚类中心会根据聚类中现有的对象被重新计算。这个过程将不断重复直到满足某个终止条件。终止条件可以是以下任何一个：

1.  没有（或最小数目）对象被重新分配给不同的聚类。
2.  没有（或最小数目）聚类中心再发生变化。
3.  误差平方和局部最小。

# Spark实现K-Means #

## Spark机器学习库 ##

Spark 机器学习库从 1.2 版本以后被分为两个包，分别是：

*  spark.mllib

> Spark MLlib 历史比较长了，1.0 以前的版本中已经包含了，提供的算法实现都是基于原始的 RDD，从学习角度上来讲，其实比较容易上手。如果您已经有机器学习方面的经验，那么您只需要熟悉下 MLlib 的 API 就可以开始数据分析工作了。想要基于这个包提供的工具构建完整并且复杂的机器学习流水线是比较困难的。

*  spark.ml

> Spark ML Pipeline 从 Spark1.2 版本开始，目前已经从 Alpha 阶段毕业，成为可用并且较为稳定的新的机器学习库。ML Pipeline 弥补了原始 MLlib 库的不足，向用户提供了一个基于 DataFrame 的机器学习工作流式 API 套件，使用 ML Pipeline API，我们可以很方便的把数据处理，特征转换，正则化，以及多个机器学习算法联合起来，构建一个单一完整的机器学习流水线。显然，这种新的方式给我们提供了更灵活的方法，而且这也更符合机器学习过程的特点。

## 算法步骤 ##

1.  将数据清洗为向量形式
2.  `随机`选取初始聚类中心`K`
3.  计算其余所有点到聚类中心的距离，并把每个点划分到离它最近的聚类中心所在的聚类中去。在这里，衡量距离一般有多个函数可以选择，最常用的是欧几里得距离 (Euclidean Distance), 也叫欧式距离。公式如下： D ( C , X ) = ∑ i − 1 n ( c i − x i ) 2 D(C,X) = \\sqrt\{\\sum\_\{i-1\}^n (c\_i - x\_i)^2\} D(C,X)=i−1∑n(ci−xi)2 上式C 代表中心点，X 代表任意一个非中心点。
4.  第三步，重新计算每个聚类中所有点的平均值，并将其作为新的聚类中心点。
5.  最后，重复 (二)，(三) 步的过程，直至聚类中心不再发生改变，或者算法达到预定的迭代次数，又或聚类中心的改变小于预先设定的阀值。

## 完整代码（基于spark.ml） ##

import org.apache.log4j.{ Level, Logger}
    import org.apache.spark.ml.clustering.{ KMeans, KMeansModel}
    import org.apache.spark.ml.linalg.{ Vector, Vectors}
    import org.apache.spark.sql.SparkSession
    
    object Kmeans { 
    
    	case class model_instance(features: Vector)
    
    	def main(args: Array[String]): Unit = { 
    		// 屏蔽日志
    		Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    		Logger.getLogger("org.apache.jetty.server").setLevel(Level.OFF)
    
    		val spark = SparkSession
    			.builder()
    			.appName("K-Means")
    			.master("local[*]")
    			.getOrCreate()
    
    		import spark.implicits._
    		import org.apache.spark.sql
    
    		// 读取源数据
    		val data = spark.sparkContext.textFile("E:\\IdeaProjects\\java\\spark-ml\\src\\main\\resources\\synthetic_control.data")
    
    		// 将数据拆分转换为向量形式
    		val parsedData = data.map(row => { 
    			val strs = row.trim.split("\\s+")
    			model_instance(Vectors.dense(strs.map(_.toDouble)))
    		}).toDF()
    
    		// 聚类个数
    		val numClusters = 8
    		//创建数据模型
    		val kmeas_model: KMeansModel = new KMeans()
    			.setK(numClusters)
    			.setFeaturesCol("features")
    			.setPredictionCol("prediction")
    			.fit(parsedData)
    
    		val results = kmeas_model.transform(parsedData)
    		
    		// 获取数据中心点
    		kmeas_model.clusterCenters
    			.foreach(center => { 
    				println("聚类中心:" + center)
    			})
    
    		// 将数据划分
    		results
    			.foreach(row => { 
    				println(row(0) + " 属于类 " + row(1))
    			})
    
    		//计算误差平方和
    		printf("集合内误差平方和：%s", kmeas_model.computeCost(parsedData))
    
    		spark.stop()
    	}
    }