无法让评级工作答案

【问题标题】：Unable to get Ratings to work无法让评级工作
【发布时间】：2023-03-12 16:58:01
【问题描述】：

我对 scala 和 spark 还很陌生，但我对 Python 和其他语言有很多经验。我遇到了一个奇怪的问题，我在包 org.apache.spark.mllib.recommendation.Rating 中使用了一个函数

Zepplin notebook 停止执行我的段落时，他 encoutners 这行但是我没有收到错误消息....

val ratings = spark.sql("SELECT _c0 as user,_c1 as product,_c2 as rating FROM datasetframe").rdd.map {case sql.Row(user: Int, product: Int, rating: Int) => Rating(user.toInt, product.toInt, rating.toInt)}

我试图重置 Zepplin 并开始一个新代码，分成多个段落.. 没有任何效果。我假设我的线路有语法问题，但解释器没有报告它。

有人可以帮忙吗？！

这里是代码的开头：

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

val data = spark.read.option("header", "false").csv("/user/philippe.samson/testing4/part-m-00000")
data.createOrReplaceTempView("datasetframe")
val sqlDF = spark.sql("SELECT * FROM datasetframe")

val ratings = spark.sql("SELECT _c0 as user,_c1 as product,_c2 as rating FROM datasetframe").rdd.map {case sql.Row(user: Int, product: Int, rating: Int) => Rating(user.toInt, product.toInt, rating.toInt)}

我也愿意测试其他方法。告诉我！

【问题讨论】：

新问题，这次异常被 ALS.TRAIN 抛出
我遇到的问题是：scala.MatchError: [13468020,117,1] (of class org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema)....I调查它并注意到它必须是一个数据类型问题，但我无法弄清楚......如果有人对匹配错误有见解......请告诉我！

标签： scala apache-spark apache-spark-sql apache-spark-mllib

【解决方案1】：

大部分情况下看起来还不错，使用 Spark 2，您可以通过消除那里的额外值来尝试这样的事情，

case class Rating(name:Int, product:Int, rating:Int)

val spark:SparkSession = ???
val df = spark.read.csv("/path/to/file")
.map({
case Row(u: Int, p: Int, r:Int) => Rating(u, p, r)
})

希望这会有所帮助。干杯。

【讨论】：

谢谢！我实际上发现磁盘上有一个隐藏的日志文件并找到了日志....我最终将该行更改为该行并且它起作用了：val rating = spark.sql("SELECT _c0 as user,_c1 as product ,_c2 as rate FROM datasetframe").rdd.map{case Seq(user: Int, product: Int, rate:Double) => Rating(user.toInt, product.toInt, rate.toDouble)}

【解决方案2】：

我的问题与未来的 NaN 值有关。我用这个修复了它： predictions.select([to_null(c).alias(c) for c in predictions.columns]).na.drop() 我还必须导入“从 pyspark.sql.functions 导入 col、isnan、when、trim”

【讨论】：