Spark实现K-Means算法代码示例

K-Means算法是一种基于距离的聚类算法，采用迭代的方法，计算出K个聚类中心，把若干个点聚成K类。

MLlib实现K-Means算法的原理是，运行多个K-Means算法，每个称为run，返回最好的那个聚类的类簇中心。初始的类簇中心，可以是随机的，也可以是KMean||得来的，迭代达到一定的次数，或者所有run都收敛时，算法就结束。

用Spark实现K-Means算法，首先修改pom文件，引入机器学习MLlib包：

<dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-mllib_2.10</artifactId>

<version>1.6.0</version>

</dependency>

代码：

import org.apache.log4j.{Level,Logger}

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

import org.apache.spark.mllib.clustering.KMeans

import org.apache.spark.mllib.linalg.Vectors

object Kmeans {

def main(args:Array[String]) = {

// 屏蔽日志

Logger.getLogger("org.apache.spark").setLevel(Level.WARN)

Logger.getLogger("org.apache.jetty.server").setLevel(Level.OFF)

// 设置运行环境

val conf = new SparkConf().setAppName("K-Means").setMaster("spark://master:7077")

.setJars(Seq("E:\\\\Intellij\\\\Projects\\\\SimpleGraphX\\\\SimpleGraphX.jar"))

val sc = new SparkContext(conf)

// 装载数据集

val data = sc.textFile("hdfs://master:9000/kmeans_data.txt", 1)

val parsedData = data.map(s => Vectors.dense(s.split(" ").map(_.toDouble)))

// 将数据集聚类,2个类,20次迭代,形成数据模型

val numClusters = 2

val numIterations = 20

val model = KMeans.train(parsedData, numClusters, numIterations)

// 数据模型的中心点

println("Cluster centres:")

for(c <- model.clusterCenters) {

println(" " + c.toString)

}

// 使用误差平方之和来评估数据模型

val cost = model.computeCost(parsedData)

println("Within Set Sum of Squared Errors = " + cost)

// 使用模型测试单点数据

println("Vectors 7.3 1.5 10.9 is belong to cluster:" + model.predict(Vectors.dense("7.3 1.5 10.9".split(" ")

.map(_.toDouble))))

println("Vectors 4.2 11.2 2.7 is belong to cluster:" + model.predict(Vectors.dense("4.2 11.2 2.7".split(" ")

.map(_.toDouble))))

println("Vectors 18.0 4.5 3.8 is belong to cluster:" + model.predict(Vectors.dense("1.0 14.5 73.8".split(" ")

.map(_.toDouble))))

// 返回数据集和结果

val result = data.map {

line =>

val linevectore = Vectors.dense(line.split(" ").map(_.toDouble))

val prediction = model.predict(linevectore)

line + " " + prediction

}.collect.foreach(println)

sc.stop

}

使用textFile()方法装载数据集，获得RDD，再使用KMeans.train()方法根据RDD、K值和迭代次数得到一个KMeans模型。得到KMeans模型以后，可以判断一组数据属于哪一个类。具体方法是用Vectors.dense()方法生成一个Vector，然后用KMeans.predict()方法就可以返回属于哪一个类。

运行结果：