K-均值聚類

K-均值算法試圖將一系列樣本分割成K個不同的類簇（其中K是模型的輸入參數），其形式化的目標函數稱為類簇內的方差和（within cluster sum of squared errors，WCSS）。K-均值聚類的目的是最小化所有類簇中的方差之和。標準的K-均值算法初始化K個類中心（為每個類簇中所有樣本的平均向量），後面的過程不斷重復叠代下面兩個步驟。
(1) 將樣本分到WCSS最小的類簇中。因為方差之和為歐拉距離的平方，所以最後等價於將每個樣本分配到歐拉距離最近的類中心。
(2) 根據第一步類分配情況重新計算每個類簇的類中心。
K-均值叠代算法結束條件為達到最大的叠代次數或者收斂。收斂意味著第一步類分配之後沒有改變，因此WCSS的值也沒有改變。

數據特征提取

這裏我還是會使用之前分類模型的MovieLens數據集。

// load movie data
val movies = sc.textFile("/PATH/ml-100k/u.item")
println(movies.first)
// 1|Toy Story (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Toy%20Story%20(1995)|0|0|0|1|1|1|0|0|0|0|0|0|0|0|0|0|0|0|0

• 提取電影的題材標題

在進一步處理之前，我們先從u.genre文件中提取題材的映射關系。

val genres = sc.textFile("/PATH/ml-100k/u.genre")
genres.take(5).foreach(println)
/*
unknown|0
Action|1
Adventure|2
Animation|3
Children‘s|4
*/
val genreMap = genres.filter(!_.isEmpty).map(line => line.split("\\|")).map(array => (array(1), array(0))).collectAsMap
println(genreMap)
// Map(2 -> Adventure, 5 -> Comedy, 12 -> Musical, 15 -> Sci-Fi, 8 -> Drama, 18 -> Western, ...

val titlesAndGenres = movies.map(_.split("\\|")).map { array =>
    val genres = array.toSeq.slice(5, array.size)
    val genresAssigned = genres.zipWithIndex.filter { case (g, idx) => 
        g == "1" 
    }.map { case (g, idx) => 
        genreMap(idx.toString) 
    }
    (array(0).toInt, (array(1), genresAssigned))
}
println(titlesAndGenres.first)
// (1,(Toy Story (1995),ArrayBuffer(Animation, Children‘s, Comedy)))
 

• 訓練推薦模型

// Run ALS model to generate movie and user factors
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.Rating
val rawData = sc.textFile("/PATH/ml-100k/u.data")
val rawRatings = rawData.map(_.split("\t").take(3))
val ratings = rawRatings.map{ case Array(user, movie, rating) => Rating(user.toInt, movie.toInt, rating.toDouble) }
ratings.cache
val alsModel = ALS.train(ratings, 50, 10, 0.1)

// extract factor vectors
import org.apache.spark.mllib.linalg.Vectors
val movieFactors = alsModel.productFeatures.map { case (id, factor) => (id, Vectors.dense(factor)) }
val movieVectors = movieFactors.map(_._2)
val userFactors = alsModel.userFeatures.map { case (id, factor) => (id, Vectors.dense(factor)) }
val userVectors = userFactors.map(_._2)
 

訓練聚類模型

在MLlib中訓練K-均值的方法和其他模型類似，只要把包含訓練數據的RDD傳入KMeans對象的train方法即可。註意，因為聚類不需要標簽，所以不用LabeledPoint實例，而是使用特征向量接口，即RDD的Vector數組即可。MLlib的K-均值提供了隨機和K-means||兩種初始化方法，後者是默認初始化。因為兩種方法都是隨機選擇，所以每次模型訓練的結果都不一樣。K-均值通常不能收斂到全局最優解，所以實際應用中需要多次訓練並選擇最優的模型。MLlib提供了完成多次模型訓練的方法。經過損失函數的評估，將性能最好的一次訓練選定為最終的模型。

代碼實現中，首先需要引入必要的模塊，設置模型參數：
K（numClusters）、最大叠代次數（numIteration）和訓練次數（numRuns）。然後，對電影的系數向量運行K-均值算法。最後，在用戶相關因素的特征向量上訓練K-均值模型：

// run K-means model on movie factor vectors
import org.apache.spark.mllib.clustering.KMeans
val numClusters = 5
val numIterations = 10
val numRuns = 3
val movieClusterModel = KMeans.train(movieVectors, numClusters, numIterations, numRuns)
/*
...
14/09/02 22:16:45 INFO SparkContext: Job finished: collectAsMap at KMeans.scala:193, took 0.02043 s
14/09/02 22:16:45 INFO KMeans: Iterations took 0.300 seconds.
14/09/02 22:16:45 INFO KMeans: KMeans reached the max number of iterations: 10.
14/09/02 22:16:45 INFO KMeans: The cost for the best run is 2585.6805358546403.
...
movieClusterModel: org.apache.spark.mllib.clustering.KMeansModel = [email protected]
*/
// train user model
val userClusterModel = KMeans.train(userVectors, numClusters, numIterations, numRuns)

使用聚類模型進行預測

K-均值最小化的目標函數是樣本到其類中心的歐拉距離之和，我們便可以將“最靠近類中心”定義為最小的歐拉距離。

下面讓我們定義這個度量函數，註意引入Breeze庫（MLlib的一個依賴庫）用於線性代數和向量運算：

// define Euclidean distance function
import breeze.linalg._
import breeze.numerics.pow
def computeDistance(v1: DenseVector[Double], v2: DenseVector[Double]): Double = pow(v1 - v2, 2).sum

利用上面的函數對每個電影計算其特征向量與所屬類簇中心向量的距離:

// join titles with the factor vectors, and compute the distance of each vector from the assigned cluster center
val titlesWithFactors = titlesAndGenres.join(movieFactors)
val moviesAssigned = titlesWithFactors.map { case (id, ((title, genres), vector)) => //vector可以理解為該點的坐標向量
    val pred = movieClusterModel.predict(vector)//pred為預測出的該點所屬的聚點
    val clusterCentre = movieClusterModel.clusterCenters(pred)//clusterCentre為該pred聚點的坐標向量
    val dist = computeDistance(DenseVector(clusterCentre.toArray), DenseVector(vector.toArray))//求兩坐標的距離
    (id, title, genres.mkString(" "), pred, dist) 
}
val clusterAssignments = moviesAssigned.groupBy { case (id, title, genres, cluster, dist) => cluster }.collectAsMap//根據聚點分組

我們枚舉每個類簇並輸出距離類中心最近的前20部電影

for ( (k, v) <- clusterAssignments.toSeq.sortBy(_._1)) {
    println(s"Cluster $k:")
    val m = v.toSeq.sortBy(_._5)
    println(m.take(20).map { case (_, title, genres, _, d) => (title, genres, d) }.mkString("\n")) 
    println("=====\n")
}

• Cluster 0
  包含了很多20世紀40年代、50年代和60年代的老電影，以及一些近代的戲劇：
  
• Cluster 1
  主要是一些恐怖電影：
  
  這裏寫圖片描述
• Cluster 2
  有相當一部分是喜劇和戲劇電影：
  
  這裏寫圖片描述
• Cluster 3
  和戲劇相關：
  
  這裏寫圖片描述
• Cluster 4
  主要是動作片、驚悚片和言情片：
  
  這裏寫圖片描述

正如你看到的，我們並不能明顯看出每個類簇所表示的內容。但是，也有證據表明聚類過程會提取電影之間的屬性或者相似之處，這不是單純基於電影名稱和題材容易看出來的（比如外語片的類簇和傳統電影的類簇，等等）。如果我們有更多元數據，比如導演、演員等，便有可能從每個類簇中找到更多特征定義的細節

評估聚類模型的性能

與回歸、分類和推薦引擎等模型類似，聚類模型也有很多評價方法用於分析模型性能，以及評估模型樣本的擬合度。聚類的評估通常分為兩部分：內部評估和外部評估。內部評估表示評估過程使用訓練模型時使用的訓練數據，外部評估則使用訓練數據之外的數據。
內部評價指標WCSS（我們之前提過的K-元件的目標函數）,是使類簇內部的樣本距離盡可能接近，不同類簇的樣本相對較遠。

MLlib提供的函數computeCost可以方便地計算出給定輸入數據RDD [Vector]的WCSS。下面我們使用這個方法計算電影和用戶訓練數據的性能：

// compute the cost (WCSS) on for movie and user clustering
val movieCost = movieClusterModel.computeCost(movieVectors)
val userCost = userClusterModel.computeCost(userVectors)
println("WCSS for movies: " + movieCost)
println("WCSS for users: " + userCost)
// WCSS for movies: 2586.0777166339426
// WCSS for users: 1403.4137493396831

聚類模型參數調優

不同於以往的模型，K-均值模型只有一個可以調的參數，就是K，即類中心數目。

// cross-validation for movie clusters
val trainTestSplitMovies = movieVectors.randomSplit(Array(0.6, 0.4), 123)
val trainMovies = trainTestSplitMovies(0)
val testMovies = trainTestSplitMovies(1)
val costsMovies = Seq(2, 3, 4, 5, 10, 20).map { k => (k, KMeans.train(trainMovies, numIterations, k, numRuns).computeCost(testMovies)) }
println("Movie clustering cross-validation:")
costsMovies.foreach { case (k, cost) => println(f"WCSS for K=$k id $cost%2.2f") }
/*
Movie clustering cross-validation:
WCSS for K=2 id 942.06
WCSS for K=3 id 942.67
WCSS for K=4 id 950.35
WCSS for K=5 id 948.20
WCSS for K=10 id 943.26
WCSS for K=20 id 947.10
*/

// cross-validation for user clusters
val trainTestSplitUsers = userVectors.randomSplit(Array(0.6, 0.4), 123)
val trainUsers = trainTestSplitUsers(0)
val testUsers = trainTestSplitUsers(1)
val costsUsers = Seq(2, 3, 4, 5, 10, 20).map { k => (k, KMeans.train(trainUsers, numIterations, k, numRuns).computeCost(testUsers)) }
println("User clustering cross-validation:")
costsUsers.foreach { case (k, cost) => println(f"WCSS for K=$k id $cost%2.2f") }
/*
User clustering cross-validation:
WCSS for K=2 id 544.02
WCSS for K=3 id 542.18
WCSS for K=4 id 542.38
WCSS for K=5 id 542.33
WCSS for K=10 id 539.68
WCSS for K=20 id 541.21
*/

從結果可以看出，隨著類中心數目增加，WCSS值會出現下降，然後又開始增大。另外一個現象，K-均值在交叉驗證的情況，WCSS隨著K的增大持續減小，但是達到某個值後，下降的速率突然會變得很平緩。這時的K通常為最優的K值（這稱為拐點）。

http://www.jianshu.com/p/d1b4c9f4844f

K-均值聚類——電影類型