scala.MatchError：火花 RDD 上为空

Question

我对 spark 和 scala 都比较陌生。 我试图在 spark 上使用 scala 实现协同过滤。 下面是代码

import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.Rating

val data = sc.textFile("/user/amohammed/CB/input-cb.txt")

val distinctUsers = data.map(x => x.split(",")(0)).distinct().map(x => x.toInt)

val distinctKeywords = data.map(x => x.split(",")(1)).distinct().map(x => x.toInt)

val ratings = data.map(_.split(',') match {
  case Array(user, item, rate) => Rating(user.toInt,item.toInt, rate.toDouble)
})

val model = ALS.train(ratings, 1, 20, 0.01)

val keywords = distinctKeywords collect
  distinctUsers.map(x => {(x, keywords.map(y => model.predict(x,y)))}).collect()

它抛出一个 scala.MatchError: null org.apache.spark.rdd.PairRDDFunctions.lookup(PairRDDFunctions.scala:571) 在最后一行 如果我将 distinctUsers rdd 收集到一个数组中并执行相同的代码，则 Thw 代码可以正常工作：

val users = distinctUsers collect
  users.map(x => {(x, keywords.map(y => model.predict(x, y)))})

我在处理 RDD 时哪里出错了？

Spark 版本：1.0.0 斯卡拉版本：2.10.4

Answer 1

在堆栈跟踪中进一步调用一次，MatrixFactorizationModel 源代码的第 43 行说：

val userVector = new DoubleMatrix(userFeatures.lookup(user).head)

注意model的userFeatures字段本身就是另一个RDD；我相信当匿名功能块在model 上关闭时它没有被正确序列化，因此它上面的lookup 方法失败了。我还尝试将model 和keywords 都放入广播变量中，但这也不起作用。

与其回退到 Scala 集合并失去 Spark 的优势，不如坚持使用 RDD 并利用其他转换它们的方法。

我会从这个开始：

val ratings = data.map(_.split(',') match {
  case Array(user, keyword, rate) => Rating(user.toInt, keyword.toInt, rate.toDouble)
})

// instead of parsing the original RDD's strings three separate times,
// you can map the "user" and "product" fields of the Rating case class

val distinctUsers = ratings.map(_.user).distinct()
val distinctKeywords = ratings.map(_.product).distinct()

val model = ALS.train(ratings, 1, 20, 0.01)

然后，我们不需要逐个计算每个预测，而是将所有可能的用户-关键字对的笛卡尔积作为 RDD，并使用 MatrixFactorizationModel 中的另一个 predict 方法，该方法将此类对的 RDD 作为其论据。

val userKeywords = distinctUsers.cartesian(distinctKeywords)

val predictions = model.predict(userKeywords).map { case Rating(user, keyword, rate) =>
  (user, Map(keyword -> rate))
}.reduceByKey { _ ++ _ }

现在predictions 为每个用户提供了一个不可变的地图，可以查询特定关键字的预测评级。如果您特别想要原始示例中的数组，您可以这样做：

val keywords = distinctKeywords.collect() // add .sorted if you want them in order
val predictionArrays = predictions.mapValues(keywords.map(_))

警告：我使用 Spark 1.0.1 进行了测试，因为它是我安装的，但它也应该适用于 1.0.0。