scala – RDD转换和操作只能由驱动程序调用

2024年2月5日 209次阅读

错误：

org.apache.spark.SparkException：RDD转换和操作只能由驱动程序调用,而不能在其他转换内部调用;例如,rdd1.map(x => rdd2.values.count()* x)无效,因为无法在rdd1.map转换内执行值转换和计数操作.有关更多信息,请参阅SPARK-5063.

def computeRatio(model: MatrixFactorizationModel, test_data: org.apache.spark.rdd.RDD[Rating]): Double = {
  val numDistinctUsers = test_data.map(x => x.user).distinct().count()
  val userRecs: RDD[(Int, Set[Int], Set[Int])] = test_data.groupBy(testUser => testUser.user).map(u => {
    (u._1, u._2.map(p => p.product).toSet, model.recommendProducts(u._1, 20).map(prec => prec.product).toSet)
  })
  val hitsAndMiss: RDD[(Int, Double)] = userRecs.map(x => (x._1, x._2.intersect(x._3).size.toDouble))

  val hits = hitsAndMiss.map(x => x._2).sum() / numDistinctUsers

  return hits
}

我正在使用MatrixFactorizationModel.scala中的方法,我必须映射用户,然后调用方法来获取每个用户的结果.通过这样做我引入嵌套映射,我相信导致问题：

我知道这个问题实际发生在：

val userRecs: RDD[(Int, Set[Int], Set[Int])] = test_data.groupBy(testUser => testUser.user).map(u => {
  (u._1, u._2.map(p => p.product).toSet, model.recommendProducts(u._1, 20).map(prec => prec.product).toSet)
})

因为映射时我正在调用model.recommendProducts

最佳答案 MatrixFactorizationModel是一个分布式模型,因此您不能简单地从动作或转换中调用它.你在这里做的最接近的事情是这样的：

import org.apache.spark.rdd.RDD
import org.apache.spark.mllib.recommendation.{MatrixFactorizationModel, Rating}

def computeRatio(model: MatrixFactorizationModel, testUsers: RDD[Rating]) = {
  val testData = testUsers.map(r => (r.user, r.product)).groupByKey
  val n = testData.count

  val recommendations = model
     .recommendProductsForUsers(20)
     .mapValues(_.map(r => r.product))

  val hits = testData
    .join(recommendations)
    .values
    .map{case (xs, ys) => xs.toSet.intersect(ys.toSet).size}
    .sum

  hits / n
}

笔记：

> distinct是一项昂贵的操作,完全过时,因为您可以从分组数据中获取相同的信息
>而不是groupBy,然后是投影(map),先是项目,后面是项目组.如果您只想要产品ID,则没有理由转移完整评级.

点赞