Spark ALS模型广播

Question

我对Spark的肌萎缩侧索硬化症推荐有问题。我想使用以下代码来预测我系统中的每个用户的产品

users = ... # RDD definition here
als_model = ... # trained ALS model
recommendations = [als_model.recommendProducts(u, 5) for u in users.collect()]

这段代码的问题在于，它并没有真正并行化。我更喜欢像这样的东西

recommendations = users.map(lambda u: als_model.recommendProducts(u, 5))

不幸的是，这是不可能的，因为我不能在Spark操作中使用als_model。

对我来说，另一种选择是执行

als_model.recommendProductsForUsers(1000)

这将返回所有用户的前1000个产品(这就是我试图实现的目标;))但是这个方法消耗了大量内存，最后由于堆空间溢出而导致GC错误。

有谁知道如何尽可能高效地解决这个问题？

Answer 1

摘自“Spark中的高级分析”一书，该书主要基于Spark 1.3 (第56页)：

目前，Spark MLlib的ALS实现并不支持向所有用户推荐的方法。可以一次向一个用户推荐，尽管每个用户都会启动一个需要几秒钟的短暂分布式作业。

还有一个100个用户的例子，它使用了你想要实现的相同的循环，所以目前没有更好的方法来实现它。

还要注意，recommendProducts(...)仍然以分布式方式运行，因此理论上您在这里损失的性能是因为作业非常短，并且提交和收集非常频繁。因此，一个大的工作比更多的小工作需要更少的协调。