Paste_Image.png

1.聚類算法

聚類（Cluster analysis）有時(shí)也被翻譯為簇類，其核心任務(wù)是：將一組目標(biāo)object劃分為若干個(gè)簇，每個(gè)簇之間的object盡可能相似，簇與簇之間的object盡可能相異。聚類算法是機(jī)器學(xué)習(xí)（或者說是數(shù)據(jù)挖掘更合適）中重要的一部分，除了最為簡(jiǎn)單的K-Means聚類算法外，比較常見的還有層次法（CURE、CHAMELEON等）、網(wǎng)格算法（STING、WaveCluster等），等等。
較權(quán)威的聚類問題定義：所謂聚類問題，就是給定一個(gè)元素集合D，其中每個(gè)元素具有n個(gè)可觀察屬性，使用某種算法將D劃分成k個(gè)子集，要求每個(gè)子集內(nèi)部的元素之間相異度盡可能低，而不同子集的元素相異度盡可能高。其中每個(gè)子集叫做一個(gè)簇。
K-means聚類屬于無監(jiān)督學(xué)習(xí)，以往的回歸、樸素貝葉斯、SVM等都是有類別標(biāo)簽y的，也就是說樣例中已經(jīng)給出了樣例的分類。而聚類的樣本中卻沒有給定y，只有特征x，比如假設(shè)宇宙中的星星可以表示成三維空間中的點(diǎn)集。聚類的目的是找到每個(gè)樣本x潛在的類別y，并將同類別y的樣本x放在一起。比如上面的星星，聚類后結(jié)果是一個(gè)個(gè)星團(tuán)，星團(tuán)里面的點(diǎn)相互距離比較近，星團(tuán)間的星星距離就比較遠(yuǎn)了。
與分類不同，分類是示例式學(xué)習(xí)，要求分類前明確各個(gè)類別，并斷言每個(gè)元素映射到一個(gè)類別。而聚類是觀察式學(xué)習(xí)，在聚類前可以不知道類別甚至不給定類別數(shù)量，是無監(jiān)督學(xué)習(xí)的一種。目前聚類廣泛應(yīng)用于統(tǒng)計(jì)學(xué)、生物學(xué)、數(shù)據(jù)庫技術(shù)和市場(chǎng)營(yíng)銷等領(lǐng)域，相應(yīng)的算法也非常多。

MLlib K-Means 的實(shí)現(xiàn)中包含一個(gè) k-means++ 方法的并行化變體 kmeans|| 。
? MLlib 里面的實(shí)現(xiàn)有如下的參數(shù)：
– k 是所需的類簇的個(gè)數(shù)。– maxIterations 是最大的迭代次數(shù)。
– initializationMode 這個(gè)參數(shù)決定了是用隨機(jī)初始化還是通過 k-means|| 進(jìn)行初始化。
– runs 是跑 k-means 算法的次數(shù)（ k-mean 算法不能保證能找出最優(yōu)解，如果在給定的數(shù)據(jù)集上運(yùn)行多次，算法將會(huì)返回最佳的結(jié)果）。– initializiationSteps 決定了 k-means|| 算法的步數(shù)。
– epsilon 決定了判斷 k-means 是否收斂的距離閥值。

實(shí)例介紹

在該實(shí)例中將介紹K-Means算法，K-Means屬于基于平方誤差的迭代重分配聚類算法，其核心思想十分簡(jiǎn)單：
l隨機(jī)選擇K個(gè)中心點(diǎn)；
l計(jì)算所有點(diǎn)到這K個(gè)中心點(diǎn)的距離，選擇距離最近的中心點(diǎn)為其所在的簇；
l簡(jiǎn)單地采用算術(shù)平均數(shù)（mean）來重新計(jì)算K個(gè)簇的中心；
l重復(fù)步驟2和3，直至簇類不再發(fā)生變化或者達(dá)到最大迭代值；
l輸出結(jié)果。

K-Means算法的結(jié)果好壞依賴于對(duì)初始聚類中心的選擇，容易陷入局部最優(yōu)解，對(duì)K值的選擇沒有準(zhǔn)則可依循，對(duì)異常數(shù)據(jù)較為敏感，只能處理數(shù)值屬性的數(shù)據(jù)，聚類結(jié)構(gòu)可能不平衡。

本實(shí)例中進(jìn)行如下步驟：
1.裝載數(shù)據(jù)，數(shù)據(jù)以文本文件方式進(jìn)行存放；
2.將數(shù)據(jù)集聚類，設(shè)置2個(gè)類和20次迭代，進(jìn)行模型訓(xùn)練形成數(shù)據(jù)模型；
3.打印數(shù)據(jù)模型的中心點(diǎn)；
4.使用誤差平方之和來評(píng)估數(shù)據(jù)模型；
5.使用模型測(cè)試單點(diǎn)數(shù)據(jù)；
6.交叉評(píng)估1，返回結(jié)果；交叉評(píng)估2，返回?cái)?shù)據(jù)集和結(jié)果

//測(cè)試數(shù)據(jù)
0.0 0.0 0.0
0.1 0.1 0.1
0.2 0.2 0.2
9.0 9.0 9.0
9.1 9.1 9.1
9.2 9.2 9.2

//代碼
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
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.eclipse.jetty.server").setLevel(Level.OFF)
 
    // 設(shè)置運(yùn)行環(huán)境
    val conf = new SparkConf().setAppName("Kmeans").setMaster("local[4]")
    val sc = new SparkContext(conf)
 
    // 裝載數(shù)據(jù)集
    val data = sc.textFile("/home/hadoop/upload/class8/kmeans_data.txt", 1)
    val parsedData = data.map(s => Vectors.dense(s.split(' ').map(_.toDouble)))
 
    // 將數(shù)據(jù)集聚類，2個(gè)類，20次迭代，進(jìn)行模型訓(xùn)練形成數(shù)據(jù)模型
    val numClusters = 2
    val numIterations = 20
    val model = KMeans.train(parsedData, numClusters, numIterations)
 
    // 打印數(shù)據(jù)模型的中心點(diǎn)
    println("Cluster centers:")
    for (c <- model.clusterCenters) {
      println("  " + c.toString)
    }
 
    // 使用誤差平方之和來評(píng)估數(shù)據(jù)模型
    val cost = model.computeCost(parsedData)
    println("Within Set Sum of Squared Errors = " + cost)
 
    // 使用模型測(cè)試單點(diǎn)數(shù)據(jù)
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))))
 
    // 交叉評(píng)估1，只返回結(jié)果
    val testdata = data.map(s => Vectors.dense(s.split(' ').map(_.toDouble)))
    val result1 = model.predict(testdata)
   result1.saveAsTextFile("/home/hadoop/upload/class8/result_kmeans1")
 
    // 交叉評(píng)估2，返回?cái)?shù)據(jù)集和結(jié)果
    val result2 = data.map {
      line =>
        val linevectore = Vectors.dense(line.split(' ').map(_.toDouble))
        val prediction = model.predict(linevectore)
        line + " " + prediction
    }.saveAsTextFile("/home/hadoop/upload/class8/result_kmeans2")
 
    sc.stop()
  }
}

2.協(xié)同過濾

協(xié)同過濾（Collaborative Filtering，簡(jiǎn)稱CF，WIKI上的定義是：簡(jiǎn)單來說是利用某個(gè)興趣相投、擁有共同經(jīng)驗(yàn)之群體的喜好來推薦感興趣的資訊給使用者，個(gè)人透過合作的機(jī)制給予資訊相當(dāng)程度的回應(yīng)（如評(píng)分）并記錄下來以達(dá)到過濾的目的，進(jìn)而幫助別人篩選資訊，回應(yīng)不一定局限于特別感興趣的，特別不感興趣資訊的紀(jì)錄也相當(dāng)重要。

協(xié)同過濾常被應(yīng)用于推薦系統(tǒng)。這些技術(shù)旨在補(bǔ)充用戶—商品關(guān)聯(lián)矩陣中所缺失的部分。

MLlib 當(dāng)前支持基于模型的協(xié)同過濾，其中用戶和商品通過一小組隱性因子進(jìn)行表達(dá)，并且這些因子也用于預(yù)測(cè)缺失的元素。MLLib 使用交替最小二乘法（ALS） 來學(xué)習(xí)這些隱性因子。

用戶對(duì)物品或者信息的偏好，根據(jù)應(yīng)用本身的不同，可能包括用戶對(duì)物品的評(píng)分、用戶查看物品的記錄、用戶的購(gòu)買記錄等。其實(shí)這些用戶的偏好信息可以分為兩類：
l 顯式的用戶反饋：這類是用戶在網(wǎng)站上自然瀏覽或者使用網(wǎng)站以外，顯式地提供反饋信息，例如用戶對(duì)物品的評(píng)分或者對(duì)物品的評(píng)論。
l 隱式的用戶反饋：這類是用戶在使用網(wǎng)站是產(chǎn)生的數(shù)據(jù)，隱式地反映了用戶對(duì)物品的喜好，例如用戶購(gòu)買了某物品，用戶查看了某物品的信息，等等。
（1）顯式的用戶反饋能準(zhǔn)確地反映用戶對(duì)物品的真實(shí)喜好，但需要用戶付出額外的代價(jià)；而隱式的用戶行為，通過一些分析和處理，也能反映用戶的喜好，只是數(shù)據(jù)不是很精確，有些行為的分析存在較大的噪音。但只要選擇正確的行為特征，隱式的用戶反饋也能得到很好的效果，只是行為特征的選擇可能在不同的應(yīng)用中有很大的不同，例如在電子商務(wù)的網(wǎng)站上，購(gòu)買行為其實(shí)就是一個(gè)能很好表現(xiàn)用戶喜好的隱式反饋。
（2）推薦引擎根據(jù)不同的推薦機(jī)制可能用到數(shù)據(jù)源中的一部分，然后根據(jù)這些數(shù)據(jù)，分析出一定的規(guī)則或者直接對(duì)用戶對(duì)其他物品的喜好進(jìn)行預(yù)測(cè)計(jì)算。這樣推薦引擎可以在用戶進(jìn)入時(shí)給他推薦他可能感興趣的物品。
（3）MLlib目前支持基于協(xié)同過濾的模型，在這個(gè)模型里，用戶和產(chǎn)品被一組可以用來預(yù)測(cè)缺失項(xiàng)目的潛在因子來描述。特別是我們實(shí)現(xiàn)交替最小二乘（ALS）算法來學(xué)習(xí)這些潛在的因子，在 MLlib 中的實(shí)現(xiàn)有如下參數(shù)：
l numBlocks是用于并行化計(jì)算的分塊個(gè)數(shù)（設(shè)置為-1時(shí) 為自動(dòng)配置）；
l rank是模型中隱性因子的個(gè)數(shù)；
l iterations是迭代的次數(shù)；
l lambda是ALS 的正則化參數(shù)；
l implicitPrefs決定了是用顯性反饋ALS 的版本還是用隱性反饋數(shù)據(jù)集的版本；
l alpha是一個(gè)針對(duì)于隱性反饋 ALS 版本的參數(shù)，這個(gè)參數(shù)決定了偏好行為強(qiáng)度的基準(zhǔn)。

Paste_Image.png

使用的ratings.dat的數(shù)據(jù)樣本如下所示：
1::1193::5::978300760
1::661::3::978302109
1::914::3::978301968
1::3408::4::978300275
1::2355::5::978824291
1::1197::3::978302268
1::1287::5::978302039
1::2804::5::978300719

2.用戶信息(users.dat)
用戶信息五個(gè)字段，格式為UserID::Gender::Age::Occupation::Zip-code，分為為用戶編號(hào)：：性別：：年齡：：職業(yè)::郵編，其中各個(gè)字段說明如下：
l用戶編號(hào)范圍1~6040
l性別，其中M為男性，F(xiàn)為女性
l不同的數(shù)字代表不同的年齡范圍，如：25代表25~34歲范圍
l職業(yè)信息，在測(cè)試數(shù)據(jù)中提供了21中職業(yè)分類
l地區(qū)郵編

使用的users.dat的數(shù)據(jù)樣本如下所示：

1::F::1::10::48067
2::M::56::16::70072
3::M::25::15::55117
4::M::45::7::02460
5::M::25::20::55455
6::F::50::9::55117
7::M::35::1::06810
8::M::25::12::11413

3.電影信息(movies.dat)
電影數(shù)據(jù)分為三個(gè)字段，格式為MovieID::Title::Genres，分為為電影編號(hào)：：電影名：：電影類別，其中各個(gè)字段說明如下：
l電影編號(hào)1~3952
l由IMDB提供電影名稱，其中包括電影上映年份
l電影分類，這里使用實(shí)際分類名非編號(hào)，如：Action、Crime等

使用的movies.dat的數(shù)據(jù)樣本如下所示：

1::Toy Story (1995)::Animation|Children's|Comedy
2::Jumanji (1995)::Adventure|Children's|Fantasy
3::Grumpier Old Men (1995)::Comedy|Romance
4::Waiting to Exhale (1995)::Comedy|Drama
5::Father of the Bride Part II (1995)::Comedy
6::Heat (1995)::Action|Crime|Thriller
7::Sabrina (1995)::Comedy|Romance
8::Tom and Huck (1995)::Adventure|Children's

代碼

import java.io.File
import scala.io.Source
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.rdd._
import org.apache.spark.mllib.recommendation.{ALS, Rating, MatrixFactorizationModel}
 
object MovieLensALS {
 
  def main(args: Array[String]) {
    // 屏蔽不必要的日志顯示在終端上
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
 
    if (args.length != 2) {
      println("Usage: /path/to/spark/bin/spark-submit --driver-memory 2g --class week7.MovieLensALS " +
        "week7.jar movieLensHomeDir personalRatingsFile")
      sys.exit(1)
    }
 
    // 設(shè)置運(yùn)行環(huán)境
    val conf = new SparkConf().setAppName("MovieLensALS").setMaster("local[4]")
    val sc = new SparkContext(conf)
 
    // 裝載用戶評(píng)分，該評(píng)分由評(píng)分器生成
    val myRatings = loadRatings(args(1))
    val myRatingsRDD = sc.parallelize(myRatings, 1)
 
    // 樣本數(shù)據(jù)目錄
    val movieLensHomeDir = args(0)
 
    // 裝載樣本評(píng)分?jǐn)?shù)據(jù)，其中最后一列Timestamp取除10的余數(shù)作為key，Rating為值,即(Int,Rating)
    val ratings = sc.textFile(new File(movieLensHomeDir, "ratings.dat").toString).map { line =>
      val fields = line.split("::")
      (fields(3).toLong % 10, Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble))
    }
 
    // 裝載電影目錄對(duì)照表（電影ID->電影標(biāo)題）
    val movies = sc.textFile(new File(movieLensHomeDir, "movies.dat").toString).map { line =>
      val fields = line.split("::")
      (fields(0).toInt, fields(1))
    }.collect().toMap
 
    val numRatings = ratings.count()
    val numUsers = ratings.map(_._2.user).distinct().count()
    val numMovies = ratings.map(_._2.product).distinct().count()
 
    println("Got " + numRatings + " ratings from " + numUsers + " users on " + numMovies + " movies.")
 
    // 將樣本評(píng)分表以key值切分成3個(gè)部分，分別用于訓(xùn)練 (60%，并加入用戶評(píng)分), 校驗(yàn) (20%), and 測(cè)試 (20%)
    // 該數(shù)據(jù)在計(jì)算過程中要多次應(yīng)用到，所以cache到內(nèi)存
    val numPartitions = 4
    val training = ratings.filter(x => x._1 < 6)
      .values
      .union(myRatingsRDD) //注意ratings是(Int,Rating)，取value即可
      .repartition(numPartitions)
      .cache()
    val validation = ratings.filter(x => x._1 >= 6 && x._1 < 8)
      .values
      .repartition(numPartitions)
      .cache()
    val test = ratings.filter(x => x._1 >= 8).values.cache()
 
    val numTraining = training.count()
    val numValidation = validation.count()
    val numTest = test.count()
 
    println("Training: " + numTraining + ", validation: " + numValidation + ", test: " + numTest)
 
    // 訓(xùn)練不同參數(shù)下的模型，并在校驗(yàn)集中驗(yàn)證，獲取最佳參數(shù)下的模型
    val ranks = List(8, 12)
    val lambdas = List(0.1, 10.0)
    val numIters = List(10, 20)
    var bestModel: Option[MatrixFactorizationModel] = None
    var bestValidationRmse = Double.MaxValue
    var bestRank = 0
    var bestLambda = -1.0
    var bestNumIter = -1
    for (rank <- ranks; lambda <- lambdas; numIter <- numIters) {
      val model = ALS.train(training, rank, numIter, lambda)
      val validationRmse = computeRmse(model, validation, numValidation)
      println("RMSE (validation) = " + validationRmse + " for the model trained with rank = "
        + rank + ", lambda = " + lambda + ", and numIter = " + numIter + ".")
      if (validationRmse < bestValidationRmse) {
        bestModel = Some(model)
        bestValidationRmse = validationRmse
        bestRank = rank
        bestLambda = lambda
        bestNumIter = numIter
      }
    }
 
    // 用最佳模型預(yù)測(cè)測(cè)試集的評(píng)分，并計(jì)算和實(shí)際評(píng)分之間的均方根誤差
    val testRmse = computeRmse(bestModel.get, test, numTest)
 
    println("The best model was trained with rank = " + bestRank + " and lambda = " + bestLambda  + ", and numIter = " + bestNumIter + ", and its RMSE on the test set is " + testRmse + ".")
 
    // create a naive baseline and compare it with the best model
    val meanRating = training.union(validation).map(_.rating).mean
    val baselineRmse =
      math.sqrt(test.map(x => (meanRating - x.rating) * (meanRating - x.rating)).mean)
    val improvement = (baselineRmse - testRmse) / baselineRmse * 100
    println("The best model improves the baseline by " + "%1.2f".format(improvement) + "%.")
 
    // 推薦前十部最感興趣的電影，注意要剔除用戶已經(jīng)評(píng)分的電影
    val myRatedMovieIds = myRatings.map(_.product).toSet
    val candidates = sc.parallelize(movies.keys.filter(!myRatedMovieIds.contains(_)).toSeq)
    val recommendations = bestModel.get
      .predict(candidates.map((0, _)))
      .collect()
      .sortBy(-_.rating)
      .take(10)
 
    var i = 1
    println("Movies recommended for you:")
    recommendations.foreach { r =>
      println("%2d".format(i) + ": " + movies(r.product))
      i += 1
    }
 
  sc.stop()
  }
 
  /** 校驗(yàn)集預(yù)測(cè)數(shù)據(jù)和實(shí)際數(shù)據(jù)之間的均方根誤差 **/
  def computeRmse(model: MatrixFactorizationModel, data: RDD[Rating], n: Long): Double = {
    val predictions: RDD[Rating] = model.predict(data.map(x => (x.user, x.product)))
    val predictionsAndRatings = predictions.map(x => ((x.user, x.product), x.rating))
      .join(data.map(x => ((x.user, x.product), x.rating)))
      .values
    math.sqrt(predictionsAndRatings.map(x => (x._1 - x._2) * (x._1 - x._2)).reduce(_ + _) / n)
  }
 
  /** 裝載用戶評(píng)分文件 **/
  def loadRatings(path: String): Seq[Rating] = {
    val lines = Source.fromFile(path).getLines()
    val ratings = lines.map { line =>
      val fields = line.split("::")
      Rating(fields(0).toInt, fields(1).toInt, fields(2).toDouble)
    }.filter(_.rating > 0.0)
    if (ratings.isEmpty) {
      sys.error("No ratings provided.")
    } else {
      ratings.toSeq
    }
  }
}

5 IDEA執(zhí)行情況
第一步 使用如下命令啟動(dòng)Spark集群
$cd /app/hadoop/spark-1.1.0
$sbin/start-all.sh
第二步 進(jìn)行用戶評(píng)分，生成用戶樣本數(shù)據(jù)
由于該程序中最終推薦給用戶十部電影，這需要用戶提供對(duì)樣本電影數(shù)據(jù)的評(píng)分，然后根據(jù)生成的最佳模型獲取當(dāng)前用戶推薦電影。用戶可以使用/home/hadoop/upload/class8/movielens/bin/rateMovies程序進(jìn)行評(píng)分，最終生成personalRatings.txt文件：

Paste_Image.png

*RMSE (validation) = 0.8680885498009973 for the model trained with rank = 8, lambda = 0.1, and numIter = 10.*
*RMSE (validation) = 0.868882967482595 for the model trained with rank = 8, lambda = 0.1, and numIter = 20.*
*RMSE (validation) = 3.7558695311242833 for the model trained with rank = 8, lambda = 10.0, and numIter = 10.*
*RMSE (validation) = 3.7558695311242833 for the model trained with rank = 8, lambda = 10.0, and numIter = 20.*
*RMSE (validation) = 0.8663942501841964 for the model trained with rank = 12, lambda = 0.1, and numIter = 10.*
*RMSE (validation) = 0.8674684744165418 for the model trained with rank = 12, lambda = 0.1, and numIter = 20.*
*RMSE (validation) = 3.7558695311242833 for the model trained with rank = 12, lambda = 10.0, and numIter = 10.*
*RMSE (validation) = 3.7558695311242833 for the model trained with rank = 12, lambda = 10.0, and numIter = 20.*
*The best model was trained with rank = 12 and lambda = 0.1, and numIter = 10, and its RMSE on the test set is 0.8652326018300565.*
*The best model improves the baseline by 22.30%.*

l 利用前面獲取的最佳模型，結(jié)合用戶提供的樣本數(shù)據(jù)，最終推薦給用戶如下影片

*Movies recommended for you:*
* 1: Bewegte Mann, Der (1994)*
* 2: Chushingura (1962)*
* 3: Love Serenade (1996)*
* 4: For All Mankind (1989)*
* 5: Vie est belle, La (Life is Rosey) (1987)*
* 6: Bandits (1997)*
* 7: King of Masks, The (Bian Lian) (1996)*
* 8: I'm the One That I Want (2000)*
* 9: Big Trees, The (1952)*
*10: First Love, Last Rites (1997)*