Spark機器學習(10)：ALS交替最小二乘算法

1. Alternating Least Square

ALS(Alternating Least Square)，交替最小二乘法。在機器學習中，特指使用最小二乘法的一種協同推薦算法。以下圖所示，u表示用戶，v表示商品，用戶給商品打分，可是並非每個用戶都會給每一種商品打分。好比用戶u6就沒有給商品v3打分，須要咱們推斷出來，這就是機器學習的任務。

因爲並非每一個用戶給每種商品都打了分，能夠假設ALS矩陣是低秩的，即一個m*n的矩陣，是由m*k和k*n兩個矩陣相乘獲得的，其中k<<m,n。

A_m×n=U_m×k×V_k×n

這種假設是合理的，由於用戶和商品都包含了一些低維度的隱藏特徵，好比咱們只要知道某我的喜歡碳酸飲料，就能夠推斷出他喜歡百世可樂、可口可樂、芬達，而不須要明確指出他喜歡這三種飲料。這裏的碳酸飲料就至關於一個隱藏特徵。上面的公式中，U_m×k表示用戶對隱藏特徵的偏好，V_k×n表示產品包含隱藏特徵的程度。機器學習的任務就是求出U_m×k和V_k×n。可知u_i^Tv_j是用戶i對商品j的偏好，使用Frobenius範數來量化重構U和V產生的偏差。因爲矩陣中不少地方都是空白的，即用戶沒有對商品打分，對於這種狀況咱們就不用計算未知元了，只計算觀察到的(用戶,商品)集合R。

這樣就將協同推薦問題轉換成了一個優化問題。目標函數中U和V相互耦合，這就須要使用交替二乘算法。即先假設U的初始值_U⁽⁰⁾，這樣就將問題轉化成了一個最小二乘問題，能夠根據_U⁽⁰⁾能夠計算出_V⁽⁰⁾，再根據_V⁽⁰⁾計算出_U⁽¹⁾，這樣迭代下去，直到迭代了必定的次數，或者收斂爲止。雖然不能保證收斂的全局最優解，可是影響不大。

2. MLlib的ALS實現

MLlib的ALS採用了數據分區結構，即將U分解成u₁,u₂,u₃,...u_m，V分解成v₁,v₂,v₃,...v_n，相關的u和v存放在同一個分區，從而減小分區間數據交換的成本。好比經過U計算V時，存儲u的分區是P₁,P₂...，存儲v的分區是Q₁,Q₂...，須要將不一樣的u發送給不一樣的Q，存放這個關係的塊稱做OutBlock；在P中，計算v時須要哪些u，存放這個關係的塊稱做InBlock。

好比R中有a₁₂,a₁₃,a₁₅，u₁存放在P₁，v₂,v₃存放在Q₂，v₅存放在Q₃，則須要將P₁中的u₁發送給Q₂和Q₃，這個信息存儲在OutBlock；R中有a₁₂,a₃₂，所以計算v₂須要u₁和u₃，這個信息存儲在InBlock。

直接上代碼：

import org.apache.log4j.{ Level, Logger }
import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.Rating

/**
  * Created by Administrator on 2017/7/19.
  */
object ALSTest01 {

  def main(args:Array[String]) ={
    // 設置運行環境
    val conf = new SparkConf().setAppName("ALS 01")
      .setMaster("spark://master:7077").setJars(Seq("E:\\Intellij\\Projects\\MachineLearning\\MachineLearning.jar"))
    val sc = new SparkContext(conf)
    Logger.getRootLogger.setLevel(Level.WARN)

    // 讀取樣本數據並解析
    val dataRDD = sc.textFile("hdfs://master:9000/ml/data/test.data")
    val ratingRDD = dataRDD.map(_.split(',') match {
      case Array(user, item, rate) =>
        Rating(user.toInt, item.toInt, rate.toDouble)
    })

    // 拆分紅訓練集和測試集
    val dataParts = ratingRDD.randomSplit(Array(0.8, 0.2))
    val trainingRDD = dataParts(0)
    val testRDD = dataParts(1)

    // 創建ALS交替最小二乘算法模型並訓練
    val rank = 10
    val numIterations = 10
    val alsModel = ALS.train(trainingRDD, rank, numIterations, 0.01)

    // 預測
    val user_product = trainingRDD.map {
      case Rating(user, product, rate) =>
        (user, product)
    }
    val predictions =
      alsModel.predict(user_product).map {
        case Rating(user, product, rate) =>
          ((user, product), rate)
      }

    val ratesAndPredictions = trainingRDD.map {
      case Rating(user, product, rate) =>
        ((user, product), rate)
    }.join(predictions)

    val MSE = ratesAndPredictions.map {
      case ((user, product), (r1, r2)) =>
        val err = (r1 - r2)
        err * err
    }.mean()

    println("Mean Squared Error = " + MSE)

    println("User" + "\t" + "Products" + "\t" + "Rate" + "\t" + "Prediction")
    ratesAndPredictions.collect.foreach(
      rating => {
        println(rating._1._1 + "\t" + rating._1._2 + "\t" + rating._2._1 + "\t" + rating._2._2)
      }
    )

  }

}

其中ALS.train()函數的4個參數分別是訓練用的數據集，特徵數量，迭代次數，和正則因子。

運行結果：

可見，預測結果仍是很是準確的。