Spark MLib ALS输入输出域

Question

我正在使用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()时，输入/输出域是什么？“收视率”是不是在5分的范围内呢？这在任何地方都没有文档记录。

Answer 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}")