я получаю сообщение об ошибке: значение toDF не является членом org.apache.spark.rdd.RDD

Question

Я написал "import sqlContext.implicits._";Однако это все еще не работает.Это верно в искровой оболочке.Почему это не правильно в этой ситуации?Я видел много других методов для перевода rdd в фрейм данных, но большая часть моего кода была написана как toDF ().Как сделать, чтобы toDF работал?ошибка:

import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.ml.tuning.{ParamGridBuilder, CrossValidator}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._
import sys.process._

"rm -f ./ml-1m.zip".!
"wget http://files.grouplens.org/datasets/movielens/ml-1m.zip".!

"ls ./ml-1m.zip".!

"rm -r ./ml-1m".!
"unzip ml-1m.zip".!

"ls ./ml-1m".!

val ratings_raw = sc.textFile("./ml-1m/ratings.dat")
ratings_raw.takeSample(false,10, seed=0).foreach(println)

case class Rating(userId: Int, movieId: Int, rating: Float)
val ratings = ratings_raw.map(x => x.split("::")).map(r => Rating(r(0).toInt, r(1).toInt, r(2).toFloat)).toDF().na.drop()

Answer 1

Я попробовал ваш код, он отлично работает!

Однако я использовал сеанс зажигания, как показано ниже

val spark = SparkSession.builder
            .master("local")
            .appName("test1")
            .getOrCreate()

Вместо устаревших

val sqlContext = new org.apache.spark.sql.SQLContext(sc)

Answer 2

если вы находитесь в spark-shell, вам не нужно создавать новый SQLContext

val sqlContext = new org.apache.spark.sql.SQLContext(sc)

, вы можете использовать spark напрямую

scala> import spark.implicits._

scala> val ratings_raw = sc.textFile("./ml-1m/ratings.dat")
ratings_raw: org.apache.spark.rdd.RDD[String] = ./ml-1m/ratings.dat MapPartitionsRDD[1] at textFile at <console>:38

scala> case class Rating(userId: Int, movieId: Int, rating: Float)
defined class Rating

scala> val ratings = ratings_raw.map(x => x.split("::")).map(r => Rating(r(0).toInt, r(1).toInt, r(2).toFloat)).toDF().na.drop()
ratings: org.apache.spark.sql.DataFrame = [userId: int, movieId: int ... 1 more field]

scala> ratings
res3: org.apache.spark.sql.DataFrame = [userId: int, movieId: int ... 1 more field]

scala> ratings.printSchema
root
 |-- userId: integer (nullable = false)
 |-- movieId: integer (nullable = false)
 |-- rating: float (nullable = false)