Spark 2.0 ALS Рекомендация, как рекомендовать пользователю

Я следовал руководству, указанному в ссылке http://ampcamp.berkeley.edu/big-data-mini-course/movie-recommendation-with-mllib.html

Но это устарело, так как использует подход Spark Mlib RDD. В New Spark 2.0 используется подход DataFrame. Теперь моя проблема в том, что у меня есть обновленный код

val ratings = spark.read.textFile("data/mllib/als/sample_movielens_ratings.txt")
  .map(parseRating)
  .toDF()
val Array(training, test) = ratings.randomSplit(Array(0.8, 0.2))

// Build the recommendation model using ALS on the training data
val als = new ALS()
  .setMaxIter(5)
  .setRegParam(0.01)
  .setUserCol("userId")
  .setItemCol("movieId")
  .setRatingCol("rating")
val model = als.fit(training)
// Evaluate the model by computing the RMSE on the test data
val predictions = model.transform(test)

Теперь вот проблема. В старом коде полученная модель была MatrixFactorizationModel, теперь у нее есть собственная модель (ALSModel).

В MatrixFactorizationModel вы можете напрямую сделать

val recommendations = bestModel.get
  .predict(userID)

Что даст список продуктов с наибольшей вероятностью того, что они понравятся пользователю.

Но теперь нет метода .predict. Любая идея, как рекомендовать список продуктов с учетом идентификатора пользователя


machine-learning apache-spark scala apache-spark-2.0
person KaustubhKhati    schedule 20.12.2016    source источник

Ответы (3)


arrow_upward
6
arrow_downward

Используйте метод transform для модели:

import spark.implicits._
val dataFrameToPredict = sparkContext.parallelize(Seq((111, 222)))
    .toDF("userId", "productId")
val predictionsOfProducts = model.transform (dataFrameToPredict)

Есть тикет jira для реализации метода «рекомендовать (пользователь | продукт)», но его еще нет в ветке по умолчанию.

Теперь у вас есть DataFrame со счетом для пользователя.

Вы можете просто использовать orderBy и ограничить показ N рекомендуемых продуктов:

// where is for case when we have big DataFrame with many users
model.transform (dataFrameToPredict.where('userId === givenUserId))
    .select ('productId, 'prediction)
    .orderBy('prediction.desc)
    .limit(N)
    .map { case Row (productId: Int, prediction: Double) => (productId, prediction) }
    .collect()

DataFrame dataFrameToPredict может быть большим DataFrame пользовательского продукта, например, все пользователи x все продукты.

person T. Gawęda    schedule 20.12.2016
comment
Преобразование будет только преобразовывать данную таблицу и добавлять значения предсказания для этого конкретного пользователя и фильма, но мне нужен список, поскольку мы собираемся использовать поток, в который придет идентификатор пользователя, и мы можем порекомендовать ему продукты. Но спасибо за помощь - person KaustubhKhati; 21.12.2016
comment
@user2983451 user2983451 Добавлен код, чтобы рекомендовать продукт N для данного пользователя;) - person T. Gawęda; 21.12.2016

arrow_upward
2
arrow_downward

Модель ALS в Spark содержит следующие полезные методы:

  • recommendForAllItems(int numUsers)

    Возвращает число пользователей, рекомендуемых для каждого элемента, для всех элементов.

  • recommendForAllUsers(int numItems)

    Возвращает первые элементы numItems, рекомендуемые для каждого пользователя, для всех пользователей.

  • recommendForItemSubset(Dataset<?> dataset, int numUsers)

    Возвращает количество пользователей, рекомендуемых для каждого идентификатора элемента в наборе входных данных.

  • recommendForUserSubset(Dataset<?> dataset, int numItems)

    Возвращает верхние элементы numItems, рекомендуемые для каждого идентификатора пользователя в наборе входных данных.

например Питон

from pyspark.ml.recommendation import ALS
from pyspark.sql.functions import explode

alsEstimator = ALS()

(alsEstimator.setRank(1)
  .setUserCol("user_id")
  .setItemCol("product_id")
  .setRatingCol("rating")
  .setMaxIter(20)
  .setColdStartStrategy("drop"))

alsModel = alsEstimator.fit(productRatings)

recommendForSubsetDF = alsModel.recommendForUserSubset(TargetUsers, 40)

recommendationsDF = (recommendForSubsetDF
  .select("user_id", explode("recommendations")
  .alias("recommendation"))
  .select("user_id", "recommendation.*")
)

display(recommendationsDF)

например Скала:

import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.sql.functions.explode 

val alsEstimator = new ALS().setRank(1)
  .setUserCol("user_id")
  .setItemCol("product_id")
  .setRatingCol("rating")
  .setMaxIter(20)
  .setColdStartStrategy("drop")

val alsModel = alsEstimator.fit(productRatings)

val recommendForSubsetDF = alsModel.recommendForUserSubset(sampleTargetUsers, 40)

val recommendationsDF = recommendForSubsetDF
  .select($"user_id", explode($"recommendations").alias("recommendation"))
  .select($"user_id", $"recommendation.*")

display(recommendationsDF)
person Joshua Cook    schedule 19.09.2018

arrow_upward
0
arrow_downward

Вот что я сделал, чтобы получить рекомендации для конкретного пользователя с помощью spark.ml:

import com.github.fommil.netlib.BLAS.{getInstance => blas}

userFactors.lookup(userId).headOption.fold(Map.empty[String, Float]) { user =>

  val ratings = itemFactors.map { case (id, features) =>
    val rating = blas.sdot(features.length, user, 1, features, 1)
    (id, rating)
  }

  ratings.sortBy(_._2).take(numResults).toMap
}

И userFactors, и itemFactors в моем случае равны RDD[(String, Array[Float])], но вы сможете сделать что-то подобное с DataFrames.

person James Ward    schedule 18.01.2017