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
  }
}