我正在面对与Spark中并发相关的问题,这阻止了我在生产中使用它,但我知道有一种方法可以解决。我正在尝试在订单历史记录上为10亿个产品中的700万个用户运行Spark ALS。首先,我正在获取一个不同用户列表,然后在这些用户上运行循环以获取推荐结果,这是非常缓慢的过程,需要数天才能为所有用户获取推荐结果。我尝试使用笛卡尔积获取所有用户和产品的推荐结果,但是再次将其提供给Elasticsearch时,我必须为每个用户过滤和排序记录,然后才能将其提供给其他API进行消费。
因此,请建议一个在这种情况下非常可扩展且可用于实时推荐的解决方案。
以下是我的Scala代码片段,它将帮助您了解我当前如何解决这个问题:
因此,请建议一个在这种情况下非常可扩展且可用于实时推荐的解决方案。
以下是我的Scala代码片段,它将帮助您了解我当前如何解决这个问题:
// buy_values -> RDD with Rating(<int user_id>, <int product_id>, <double rating>)
def recommend_for_user(user: Int): Unit = {
println("Recommendations for User ID: " + user);
// Product IDs which are not bought by user
val candidates = buys_values
.filter(x => x("customer_id").toString.toInt != user)
.map(x => x("product_id").toString.toInt)
.distinct().map((user, _))
// find 30 products with top rating
val recommendations = bestModel.get
.predict(candidates)
.takeOrdered(30)(Ordering[Double].reverse.on(x => x.rating))
var i = 1
var ESMap = Map[String, String]()
recommendations.foreach { r =>
ESMap += r.product.toString -> bitem_ids.value(r.product)
}
// push to elasticsearch with user as id
client.execute {
index into "recommendation" / "items" id user fields ESMap
}.await
// remove candidate RDD from memory
candidates.unpersist()
}
// iterate on each user to get recommendations for the user [slow process]
user_ids.foreach(recommend_for_user)