scala – 加速Spark MLLib中大型数据集的协同过滤

我正在使用MLlib的矩阵分解来向用户推荐项目.我有一个很大的隐式交互矩阵,M = 2000万用户和N = 50k项目.在训练模型之后,我想获得针对每个用户的推荐的简短列表(例如,200).我在MatrixFactorizationModel中尝试过recommendedProductsForUsers,但它非常慢(跑了9个小时但距离完成还很远.我正在测试50个执行器,每个都有8g内存).这可能是预期的,因为recommendedProductsForUsers需要计算所有M * N个用户项交互并获得每个用户的顶部.

我会尝试使用更多的执行程序,但是从我在Spark UI上的应用程序细节中看到的,我怀疑它可以在几小时或一天内完成,即使我有1000个执行程序(在9小时之后它仍然在平面图中这里https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/recommendation/MatrixFactorizationModel.scala#L279-L289,总共10000个任务只有~200完成)
除了增加执行者数量之外,还有其他任何我可以调整的东西来加快推荐过程吗？

这是示例代码：

val data = input.map(r => Rating(r.getString(0).toInt,r.getString(1).toInt,r.getLong(2))).cache
val rank = 20
val alpha = 40
val maxIter = 10
val lambda = 0.05
val checkpointIterval = 5
val als = new ALS()
    .setImplicitPrefs(true)
    .setCheckpointInterval(checkpointIterval)
    .setRank(rank)
    .setAlpha(alpha)
    .setIterations(maxIter)
    .setLambda(lambda)
val model = als.run(ratings)
val recommendations = model.recommendProductsForUsers(200)
recommendations.saveAsTextFile(outdir)

javaSparkContext.setCheckpointDir("checkpoint/");

--conf spark.driver.maxResultSize=5g --conf spark.yarn.executor.memoryOverhead=4000