scala + spark：序列化任务以获取建议答案

【问题标题】：scala + spark: serialize a task for recommendationsscala + spark：序列化任务以获取建议
【发布时间】：2014-11-10 23:33:15
【问题描述】：

我一直在使用 Scala + Spark 和 Movie Recommendation with MLib 教程。

获得我的预测后，我需要每个用户的前 3 个项目。

val predictions = 
  model.predict(usersProducts).map { case Rating(user, product, rate) =>
  (user, product, rate)
}

我试过了：

def myPrint(x:(Int, Int, Double)) = println(x._1 + ":" + x._2 + " - " +x._3)

predictions.collect().sortBy(- _._3).groupBy(_._1).foreach( t2 => t2._2.take(3).foreach(myPrint) )

（_.1 是用户，_.2 是项目，_.3 是费率）

我必须添加“collect()”方法才能使其工作，但我无法序列化此任务。顺便说一句，我添加了myPrint方法，因为我不知道如何从最后一行获取集合或地图。

有什么想法让它可序列化吗？

知道从最后一行获取集合/地图吗？

如果我不能做得更好，我会在 myPrint 中写入数据库并在 1000 次插入后提交。

谢谢。

【问题讨论】：

标签： scala serialization apache-spark recommendation-engine

【解决方案1】：

您可以通过稍微修改您的方法来确保所有计算都在 RDD 中完成：

predictions.sortBy(- _.rating).groupBy(_.user)
  .flatMap(_._2.take(3)).foreach(println)

【讨论】：

【解决方案2】：

调用方法的任务必须序列化包含该方法的对象。尝试改用函数值：

val myPrint: ((Int, Int, Double)) => Unit = x => ...

您不希望在开头使用 collect()，这会破坏使用 Spark 的全部意义。

我不明白你在说什么“获取收藏/地图”。 .take(3) 已经返回了一个集合。

【讨论】：

这并没有解决我的问题。我删除了收集并使用了您的函数，但 spark 一直说任务不可序列化，但感谢有关使用函数值的提示
您的具体错误是什么？它可能来自您程序的早期阶段，因为即使您在其中有 collect()，该任务也无法序列化。
异常说的是：org.apache.spark.SparkException: Task not serializable ... 原因：java.io.NotSerializableException: org.apache.spark.SparkContext ...
所以你试图传递上下文，是的，这是行不通的。我真的不认为是您发布的代码给您带来了问题。请注意，RDD 是惰性的，因此错误可能来自几行后面的内容。

【解决方案3】：

阅读lmm的答案并做一些研究后，我这样解决了我的问题：

首先，我开始使用 Rating 对象而不是 Tuple：

val predictions = model.predict(usersProducts)

然后我定义函数值如下，现在我在这里做“取”：

def myPrint: ((Int, Iterable[Rating])) => Unit = x => x._2.take(3).foreach(println)

所以，现在我用这种方式混合所有内容：

predictions.sortBy(- _.rating).groupBy(_.user).foreach(myPrint)

【讨论】：