通过IDEA及hadoop平台实现k-means聚类算法

由于实验室任务方向变更，本文不再更新~

有段时间没有操作过，发现自己忘记一些步骤了，这篇文章会记录相关步骤，并随时进行补充修改。

1 基础步骤，即相关环境部署及数据准备

数据文件类型为.csv文件，excel直接另存为即可，以逗号为分隔符

2 IDEA编辑代码，打jar包

参考以下链接：

IntelliJ IDEA Windows下Spark开发环境部署

IDEA开发Spark的漫漫摸索（一）

IDEA开发Spark的漫漫摸索（二）

k-means聚类代码参考：

package main.scala.yang.spark

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.clustering.KMeans

object KMeansBeijing {
  def main(args: Array[String]): Unit = {
    // 屏蔽不必要的日志显示在终端上
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    // 设置运行环境
    val conf = new SparkConf().setMaster("local").setAppName("KMeansBeijing")
    val sc = new SparkContext(conf)

    // 装载数据集
    val data = sc.textFile("file:///home/hadoop/yang/USA/AUG_tag.csv", 1)
    val parsedData = data.filter(!isColumnNameLine(_)).map(line => Vectors.dense(line.split(',').map(_.toDouble))).cache()
    //
    // 将数据集聚类，7个类，20次迭代，进行模型训练形成数据模型
    val numClusters = 4
    val numIterations = 800
    val model = KMeans.train(parsedData, numClusters, numIterations)

    // 打印数据模型的中心点
    println("Cluster centers:")
    for (c <- model.clusterCenters) {
      println("  " + c.toString)
    }

    // 使用误差平方之和来评估数据模型
    val cost = model.computeCost(parsedData)
    println("Within Set Sum of Squared Errors = " + cost)

//    // 使用模型测试单点数据
//    println("Vectors 0.2 0.2 0.2 is belongs to clusters:" + model.predict(Vectors.dense("0.2 0.2 0.2".split(' ').map(_.toDouble))))
//    println("Vectors 0.25 0.25 0.25 is belongs to clusters:" + model.predict(Vectors.dense("0.25 0.25 0.25".split(' ').map(_.toDouble))))
//    println("Vectors 8 8 8 is belongs to clusters:" + model.predict(Vectors.dense("8 8 8".split(' ').map(_.toDouble))))

    // 交叉评估1，只返回结果
    val testdata = data.filter(!isColumnNameLine(_)).map(s => Vectors.dense(s.split(',').map(_.toDouble)))
    val result1 = model.predict(testdata)
    result1.saveAsTextFile("file:///home/hadoop/yang/USA/AUG/result1")

    // 交叉评估2，返回数据集和结果
    val result2 = data.filter(!isColumnNameLine(_)).map {
      line =>
        val linevectore = Vectors.dense(line.split(',').map(_.toDouble))
        val prediction = model.predict(linevectore)
        line + " " + prediction
    }.saveAsTextFile("file:///home/hadoop/yang/USA/AUG/result2")

    sc.stop()
  }

  private def isColumnNameLine(line: String): Boolean = {
    if (line != null && line.contains("Electricity")) true
    else false
  }
}

3 通过WinSCP将jar包上传到hadoop平台本地服务器上

注：直接拖拽即可

4 通过SecureCRT在hadoop平台上执行相关命令

4.1 进入spark文件夹下

4.2 通过spark-submit命令提交任务（jar包）到集群

4.3 通过WinSCP查看结果

注：4.1和4.2可以综合在一条命令中：