Spark MLlib ALS 输入和输出域答案

【问题标题】：Spark MLib ALS input and output domainsSpark MLlib ALS 输入和输出域
【发布时间】：2016-09-06 23:11:07
【问题描述】：

我正在使用 Spark MLib ALS 并尝试使用 trainImplicit() 接口将用户购买的商品数量作为隐式偏好提供给它。我不知道如何验证我的模型。我的输入在域 [1, inf) 中，但输出预测似乎在 (0, 1) 中浮动。

常用的代码：

from pyspark.mllib.recommendation import ALS, MatrixFactorizationModel, Rating
from pyspark.sql import HiveContext
from pyspark import SparkContext

sc = SparkContext(appName="Quantity Prediction Model")
hive = HiveContext(sc)

d = hive.sql("select o.user_id as user, l.product_id as product, sum(l.quantity) as qty from order_line l join order_order o ON l.order_id = o.id group by o.user_id, l.product_id")
d.write.save('user_product_qty')

ratings = d.rdd.map(tuple)
testdata = ratings.map(lambda t: (t[0], t[1]))

for rank in (4, 8, 12):
    model = ALS.trainImplicit(ratings, rank, 10, alpha=0.01)

    predictions = model.predictAll(testdata).map(lambda r: ((r[0], r[1]), r[2]))
    ratesAndPreds = ratings.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)

    # Error is pretty bad because output raitings aren't in the same domain as quantity
    ratesAndPreds = ratings.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)
    MSE = ratesAndPreds.map(lambda r: (r[1][0] - r[1][1])**2).mean()

    print("Rank: {} MSE: {}".format(rank, MSE))

额外功劳：当使用train() 时，输入/输出域是什么？ “收视率”是否会达到五分制？这在任何地方都没有记录。

【问题讨论】：

标签： pyspark apache-spark-mllib

【解决方案1】：

RMSE 不是隐式 ALS 的理想指标（原始论文提出了一种更精细的评估技术）。但是，如果您也将输入评分映射到 (-1;1)，您仍然可以应用 RMSE 来评估隐式 ALS 训练结果。

详情请见https://github.com/apache/spark/pull/597。

最后，一些代码可以帮助您入门（灵感来自 Spark 中用于 ALS 的 MovieLens 示例）：

// RMSE
logger.info(s"Calculating RMSE on ${testingSet.count()} ratings")
def groupRatings(rs: RDD[MLlibRating]): RDD[((Int, Int), Double)] =
  rs.map { r => ((r.user, r.product), r.rating) }

// When using implicit ALS we should treat actual and predicted
// ratings as confidence levels. See also apache/spark#597.
//
// Predicted ratings are clamped to [0;1]
def clampPredicted(r: Double): Double =
  math.max(math.min(r, 1.0), 0.0)

def clampActual(r: Double): Double = if (r > 0.0) 1.0 else 0.0

def sqr(x: Double): Double = x * x

val ratingSquaredErrors =
  groupRatings(alsModel.predict(testingSet.map { r => (r.user, r.product) }))
  .join(groupRatings(testingSet))
  .map { case (_, (predictedRating, actualRating)) =>
         sqr(clampPredicted(predictedRating) - clampActual(actualRating)) }
val rmse = sqrt(ratingSquaredErrors.mean())
logger.info(s"RMSE: ${rmse}")

【讨论】：