MLlib-协同过滤

协同过滤
- 显示vs隐式反馈
- 参数调整
实例
教程

协同过滤

协同过滤是推荐系统的常用方法。可以填充user-item相关矩阵中的缺失值。MLlib支持基于模型的协同过滤，即使用能够预测缺失值的一个隐藏因素集合来表示用户和产品。MLlib使用交替做小二乘法（alternating least squares, ALS）学习隐藏因子。MLlib算法中的参数如下：

numBlocks 并行计算的block数（-1为自动配置）
rank 模型中隐藏因子数
iterations 算法迭代次数
lambda ALS中的正则化参数
implicitPrefs 使用显示反馈ALS变量或隐式反馈
alpha ALS隐式反馈变化率用于控制 the baseline confidence in preference observations

显示vs隐式fankui

基于矩阵分解的协同过滤的标准方法一般将用户商品矩阵中的元素作为用户对商品的显性偏好。

真实的案例中通常只有隐式反馈（例如，查看，点击，购买，喜欢，分享等）。MLlib中处理此类数据的方法来自于Collaborative Filtering for Implicit Feedback Datasets。本质上，这个方法将数据作为二元偏好值和偏好强度的一个结合，而不是对评分矩阵直接进行建模。因此，评价就不是与用户对商品的显性评分而是和所观察到的用户偏好强度关联了起来。然后，这个模型将尝试找到隐语义因子来预估一个用户对一个商品的偏好。

参数调整

V1.1以来，在结局每个最小二乘问题中通过调整lambda参数来控制更新用户因子时的用户产生比例或在更新物品因子时接收物品的比例。这种方法叫做“ALS-WR”，来自论文“Large-Scale Parallel Collaborative Filtering for the Netflix Prize”。他使lambda对数据规模的依赖更小。所以，这里使用从抽样数据中学到的最优参数去填充应用到整个数据集并期盼会有类似表现。

实例

例子中，使用打分数据，每一行包括：用户，产品和打分。使用默认的ALS.train()方法，假定打分是显示的。使用打分预测的错误率军方法来评价推荐模型。（注：这里只调研了scala）

import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
import org.apache.spark.mllib.recommendation.Rating

// Load and parse the data
val data = sc.textFile("data/mllib/als/test.data")
val ratings = data.map(_.split(',') match { case Array(user, item, rate) =>
    Rating(user.toInt, item.toInt, rate.toDouble)
  })

// Build the recommendation model using ALS
val rank = 10
val numIterations = 10
val model = ALS.train(ratings, rank, numIterations, 0.01)

// Evaluate the model on rating data
val usersProducts = ratings.map { case Rating(user, product, rate) =>
  (user, product)
}
val predictions = 
  model.predict(usersProducts).map { case Rating(user, product, rate) => 
    ((user, product), rate)
  }
val ratesAndPreds = ratings.map { case Rating(user, product, rate) => 
  ((user, product), rate)
}.join(predictions)
val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) => 
  val err = (r1 - r2)
  err * err
}.mean()
println("Mean Squared Error = " + MSE)

// Save and load model
model.save(sc, "myModelPath")
val sameModel = MatrixFactorizationModel.load(sc, "myModelPath")

如果打分矩阵是从其他源信息得到的，可以使用 trainImplicit 方法得到更好的结果。

val alpha = 0.01
val lambda = 0.01
val model = ALS.trainImplicit(ratings, rank, numIterations, lambda, alpha)

请参照Self-Contained Applications运行上面的例子。确保build文件中包含spark-mllib依赖。

教程

The training exercises from the Spark Summit 2014 include a hands-on tutorial for personalized movie recommendation with MLlib.

使用MovieLens数据进行协同过滤

在IDEA中新建工程filter-starter，新建文件src/Filter.scala，本地运行程序：

/**
 * Created by *****qin on 2014-12-25.
 */
import org.apache.spark._
import org.apache.spark.SparkContext._
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.Rating


object Filter {
  def main(args: Array[String]) {
    //load data
    val conf = new SparkConf().setAppName("filter").setMaster("local")
    val sc = new SparkContext(conf)
    val data = sc.textFile("E:\_spark\test_data\ua_ratings.data")
    println(data.count())
    val ratings = data.map(_.split(" ") match { case Array(user, item, rate) => Rating(user.toInt, item.toInt, rate.toDouble)})


    //build the recommendation model using ALS
    val rank = 10
    val numIterations = 20
    val model = ALS.train(ratings, rank, numIterations, 0.01)

    //Evaluate the model on the rating data
    val usersProducts = ratings.map { case Rating(user, product, rate) =>
      (user, product)
    }

    val predictions =
      model.predict(usersProducts).map { case Rating(user, product, rate) =>
        ((user, product), rate)
      }
    val ratesAndPreds = ratings.map { case Rating(user, product, rate) =>
      ((user, product), rate)
    }.join(predictions)


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

    println("Mean Squared Error = " + MSE)
  }
}

右键->"Run Filter"即可

**期间遇到一个问题，程序里已经导入了import org.apache.spark._ 居然不管用，总是提示错误

Error:(36, 7) value join is not a member of org.apache.spark.rdd.RDD[((Int, Int), Double)]
possible cause: maybe a semicolon is missing before `value join'?
}.join(predictions)
^

找了半天发现需要导入import org.apache.spark.SparkContext._ ，虽然IDEA提示unused import statement，但是编译就不报错了，凌乱了，怀疑IDEA对scala的支持

集群运行，需要修改程序：

//load data
val conf = new SparkConf().setAppName("filter")
val sc = new SparkContext(conf)
val data = sc.textFile(args(0))

将程序打成jar包，上传hdfs集群提交任务

如有疑问，欢迎评论！！！！！