Spark 2.2.0 文檔中文版 Collaborative Filtering 協同過濾 JAVA推薦系統

　　協同過濾經常使用於推薦系統，這項技術旨在填補 丟失的user-item關聯矩陣 的條目，spark.ml目前支持基於模型的協同過濾（用一些丟失條目的潛在因素在描述用戶和產品）。spark.ml使用ALS（交替最小二乘法）去學習這些潛在因素。在spark.ml中的實現有如下參數：

numBlocks：塊的數量，user和item將被分紅多少塊，以並行計算。（默認10）

ranK:模型隱含因素的個數。（默認10）

maxIter：模型的最大迭代次數。（默認10）

regParam ：ALS的正則化參數。（默認1.0）

implicitPrefs ：使用顯式反饋仍是隱式反饋。（默認false，即顯式反饋）

alpha：信心權重所應達到的基準線。（默認1.0）

nonnegative ：是否使用非負數的約束。（默認false）

　　注意：基於dataFrame - API的ALS目前只支持整數型的userID和itemID，其餘數字類型也支持，可是取值範圍必須在整數以內。

顯式反饋VS隱式反饋

　　標準的基於矩陣分解的協同過濾方法對待user-item矩陣的條目項 是顯式地給出user對item的偏好,例如,用戶給電影評級。

　　而現實生活中常見案例是隻能有隱式反饋（例如：視圖，點擊鼠標，購買，喜歡，分享……）。在spark.ml中使用的方法是：對隱式反饋數據集的協同過濾。實際上，這種方法不是直接對數據矩陣進行建模，而是將數據視爲表明用戶行爲意願強度的數字（例如點擊的次數或某人累積觀看電影的時間）。而後，這些數字與觀察到的用戶偏好的置信水平相關，而不是給予項目的明確評級。 而後，該模型嘗試找到可用於預測用戶對項目的預期偏好的潛在因素。

　　Scaling of the regularization parameter（正則化參數的換算）

　　咱們經過用戶在更新用戶因素中產生的評分，或產品在更新產品因素中收到的評分來求解每一個最小二乘問題的規則化參數 regParam 。 這種方法被命名爲「 ALS-WR 」（加權正則化交替最小二乘法），並在論文「 Large-Scale Parallel Collaborative Filtering for the Netflix Prize 」中進行了討論。 它使 regParam 對數據集的規模依賴較少，所以咱們能夠將從採樣子集學到的最佳參數應用於完整數據集，並指望能有類似的表現。

　　冷啓動策略

　　在進行預測時使用ALSModel,一般遇到這樣一種狀況，訓練模型的時候，user、item在測試數據集沒有出現。這種狀況一般發生在兩個場景:

1. 在生產中,新user或item,沒有評級的歷史，或者模型沒有訓練(這是「冷啓動問題」)。
2. 在交叉驗證中,數據分爲訓練集和評價集。當簡單的隨機生成 CrossValidator或 TrainValidationSplit,它其實是很是常見的出現這樣狀況：user、item在驗證集,但不在訓練集。

　　默認狀況下,當user、item不在模型之中，spark在ALSModel.transform分配NaN預測，這在預測系統中很是的有用，由於它代表了一個新的user、item，系統能夠作一個後備性的預測。

　　然而在交叉驗證中這是不可取的，由於任何一個NaN預測會致使評估度量的NaN結果（例如當使用RegressionEvaluator時）這使得模型選擇是不可能的。

　　spark容許用戶設置coldStartStrategy參數爲「drop」，以便刪除包含NaN值的預測的DataFrame中的任何行。 而後，將根據非NaN數據計算評估度量，並將有效。 如下示例說明了此參數的用法。

注意：目前支持的冷啓動策略是「nan」（上面提到的默認行爲）和「drop」。 將來可能會支持進一步的策略。

例子

　　在如下示例中，咱們從MovieLens數據集加載評級數據，每行由用戶，電影，評級和時間戳組成。 而後，假設咱們訓練一個ALS模型，默認狀況下假定評級是明確的（implicitPrefs是false）。 咱們經過測量評級預測的均方根偏差來評估推薦模型。

有關API的更多詳細信息，請參閱ALS Java文檔。

import java.io.Serializable;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.recommendation.ALS;
import org.apache.spark.ml.recommendation.ALSModel;

public static class Rating implements Serializable {
  private int userId;
  private int movieId;
  private float rating;
  private long timestamp;

  public Rating() {}

  public Rating(int userId, int movieId, float rating, long timestamp) {
    this.userId = userId;
    this.movieId = movieId;
    this.rating = rating;
    this.timestamp = timestamp;
  }

  public int getUserId() {
    return userId;
  }

  public int getMovieId() {
    return movieId;
  }

  public float getRating() {
    return rating;
  }

  public long getTimestamp() {
    return timestamp;
  }

  public static Rating parseRating(String str) {
    String[] fields = str.split("::");
    if (fields.length != 4) {
      throw new IllegalArgumentException("Each line must contain 4 fields");
    }
    int userId = Integer.parseInt(fields[0]);
    int movieId = Integer.parseInt(fields[1]);
    float rating = Float.parseFloat(fields[2]);
    long timestamp = Long.parseLong(fields[3]);
    return new Rating(userId, movieId, rating, timestamp);
  }
}

JavaRDD<Rating> ratingsRDD = spark
  .read().textFile("data/mllib/als/sample_movielens_ratings.txt").javaRDD()
  .map(Rating::parseRating);
Dataset<Row> ratings = spark.createDataFrame(ratingsRDD, Rating.class);
Dataset<Row>[] splits = ratings.randomSplit(new double[]{0.8, 0.2});
Dataset<Row> training = splits[0];
Dataset<Row> test = splits[1];

// Build the recommendation model using ALS on the training data
ALS als = new ALS()
  .setMaxIter(5)
  .setRegParam(0.01)
  .setUserCol("userId")
  .setItemCol("movieId")
  .setRatingCol("rating");
ALSModel model = als.fit(training);

// Evaluate the model by computing the RMSE on the test data
// Note we set cold start strategy to 'drop' to ensure we don't get NaN evaluation metrics
model.setColdStartStrategy("drop");
Dataset<Row> predictions = model.transform(test);

RegressionEvaluator evaluator = new RegressionEvaluator()
  .setMetricName("rmse")
  .setLabelCol("rating")
  .setPredictionCol("prediction");
Double rmse = evaluator.evaluate(predictions);
System.out.println("Root-mean-square error = " + rmse);

// Generate top 10 movie recommendations for each user
Dataset<Row> userRecs = model.recommendForAllUsers(10);
// Generate top 10 user recommendations for each movie
Dataset<Row> movieRecs = model.recommendForAllItems(10);

　　

　　在Spark repo中的「examples / src / main / java / org / apache / spark / examples / ml / JavaALSExample.java」中查找完整示例代碼。
若是評級矩陣是從另外一個信息來源導出的（即從其餘信號推斷出來），您能夠將implicitPrefs設置爲true以得到更好的結果：

ALS als = new ALS()
  .setMaxIter(5)
  .setRegParam(0.01)
  .setImplicitPrefs(true)
  .setUserCol("userId")
  .setItemCol("movieId")
  .setRatingCol("rating");