四、协同过滤

协同过滤[Collaborative filtering]:

协同过滤(Collaborative filtering)是推荐系统的一个常用算法。这个技术的目的在于填充user-item矩阵中的缺失项。MLlib当前支持基于模型的协同过滤，在这种方法中，用户和产品通过一个小的潜在因素（latent factors）集合来描述，这个潜在因素集合可以预测缺失项。MLlib使用交替最小二乘法alternating least squares (ALS) 来学习这些潜在因素。MLlib的实现有下列参数:

numBlocks 并行计算的块数量。（默认值为-1，表示自动配置）

rank 模型中潜在因素的数量。

iterations 迭代次数。

lambda ALS中的正则化参数。

implicitPrefs 制定是否使用显示反馈ALS变体（或者说是对隐式反馈数据的一种适应）

alpha 应用于隐式数据的ALS变体，它控制的是观察到偏好的基本置信度。

显式反馈 VS隐式反馈

基于矩阵分解的协同过滤，其标准做法是将user-time矩阵中的条目看做用户对该条目的显式偏好。

在现实世界中，通常只能使用隐式的反馈（例如，查看、点击、购买、喜欢、分享等等）。MLlib中处理这种数据的方法来自 Collaborative Filtering for Implicit Feedback Datasets(针对隐式反馈的协同过滤)。这种方法将数据作为是否偏好及对应的置信度的组合来使用，而不是对评分矩阵直接建模。也就是说评分跟观察到的用户的偏好置信度相关，而不是作为对条目的显式评分。然后尝试找到潜在因素从而预测用户对某个条目的喜好。

正则化参数的扩展

从版本1.1开始，MLlib中对每个解决最小二乘问题的正则化参数lambda做了扩展：一个是在更新用户因素时用户产生的评分数量；另一个是在更新产品因素时产品被评分的数量。这个方法叫做ALS-WR(alternating-least-squares with weighted-λ - regularization)，这篇论文有详细的介绍：Large-Scale Parallel Collaborative Filtering for the Netflix Prize。该算法减小了参数lambda对数据集规模的依赖。所以我们可以把一个从抽样子集上学习到的最好的参数应用到全部数据集上，并能预计得到一样好的效果。

import scala.Tuple2;

import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.mllib.recommendation.ALS;
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
import org.apache.spark.mllib.recommendation.Rating;
import org.apache.spark.SparkConf;

SparkConf conf = new SparkConf().setAppName("Java Collaborative Filtering Example");
JavaSparkContext jsc = new JavaSparkContext(conf);

// Load and parse the data
String path = "data/mllib/als/test.data";
JavaRDD<String> data = jsc.textFile(path);
JavaRDD<Rating> ratings = data.map(
  new Function<String, Rating>() {
    public Rating call(String s) {
      String[] sarray = s.split(",");
      return new Rating(Integer.parseInt(sarray[0]), Integer.parseInt(sarray[1]),
        Double.parseDouble(sarray[2]));
    }
  });

// Build the recommendation model using ALS
int rank = 10;int numIterations = 10;
MatrixFactorizationModel model = ALS.train(JavaRDD.toRDD(ratings), rank, numIterations, 0.01);

// Evaluate the model on rating data
JavaRDD<Tuple2<Object, Object>> userProducts = ratings.map(
  new Function<Rating, Tuple2<Object, Object>>() {
    public Tuple2<Object, Object> call(Rating r) {
      return new Tuple2<Object, Object>(r.user(), r.product());
    }
  }
);
JavaPairRDD<Tuple2<Integer, Integer>, Double> predictions = JavaPairRDD.fromJavaRDD(
  model.predict(JavaRDD.toRDD(userProducts)).toJavaRDD().map(
    new Function<Rating, Tuple2<Tuple2<Integer, Integer>, Double>>() {
      public Tuple2<Tuple2<Integer, Integer>, Double> call(Rating r){
        return new Tuple2<Tuple2<Integer, Integer>, Double>(
          new Tuple2<Integer, Integer>(r.user(), r.product()), r.rating());
      }
    }
  ));
JavaRDD<Tuple2<Double, Double>> ratesAndPreds =
  JavaPairRDD.fromJavaRDD(ratings.map(
    new Function<Rating, Tuple2<Tuple2<Integer, Integer>, Double>>() {
      public Tuple2<Tuple2<Integer, Integer>, Double> call(Rating r){
        return new Tuple2<Tuple2<Integer, Integer>, Double>(
          new Tuple2<Integer, Integer>(r.user(), r.product()), r.rating());
      }
    }
  )).join(predictions).values();
double MSE = JavaDoubleRDD.fromRDD(ratesAndPreds.map(
  new Function<Tuple2<Double, Double>, Object>() {
    public Object call(Tuple2<Double, Double> pair) {
      Double err = pair._1() - pair._2();
      return err * err;
    }
  }).rdd()).mean();
System.out.println("Mean Squared Error = " + MSE);

// Save and load model
model.save(jsc.sc(), "target/tmp/myCollaborativeFilter");
MatrixFactorizationModel sameModel = MatrixFactorizationModel.load(jsc.sc(),
  "target/tmp/myCollaborativeFilter");