Scala ratings.csv文件 文件里面的内容包含了每一个用户对于每一部电影的评分。数据格式如下： userId, movieId, rating, timestamp userId: 每个用户的id movieId: 每部电影的id rating: 用户评分，是5星制 timestamp: 自1970年1月1日零点后到用户提交评价的时间的秒数 movies.csv文件 movieId: 每部电影的Id title：每部电影的名字 编程实现RDD转DataFrame，并选择一个合理的SQL分析。import org.apache.spark.sql.Row import org.apache.spark.sql.types._

好的，以下是实现RDD转DataFrame的代码：

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

// 读取ratings.csv文件
val ratingsRDD = sc.textFile("ratings.csv")
// 将每行数据按逗号分隔，并转化为Row对象
val ratingsRowRDD = ratingsRDD.map(line => {
  val fields = line.split(",")
  Row(fields(0).toInt, fields(1).toInt, fields(2).toDouble, fields(3).toLong)
})
// 定义schema，即数据类型
val ratingsSchema = StructType(
  StructField("userId", IntegerType, true) ::
  StructField("movieId", IntegerType, true) ::
  StructField("rating", DoubleType, true) ::
  StructField("timestamp", LongType, true) :: Nil
)
// 构建DataFrame
val ratingsDF = spark.createDataFrame(ratingsRowRDD, ratingsSchema)

// 读取movies.csv文件
val moviesRDD = sc.textFile("movies.csv")
// 将每行数据按逗号分隔，并转化为Row对象
val moviesRowRDD = moviesRDD.map(line => {
  val fields = line.split(",")
  Row(fields(0).toInt, fields(1))
})
// 定义schema，即数据类型
val moviesSchema = StructType(
  StructField("movieId", IntegerType, true) ::
  StructField("title", StringType, true) :: Nil
)
// 构建DataFrame
val moviesDF = spark.createDataFrame(moviesRowRDD, moviesSchema)

接下来是一个合理的SQL分析示例，假设我们想要查找电影评分最高的前10部电影：

import org.apache.spark.sql.functions._

// 计算每部电影的平均评分
val avgRatingsDF = ratingsDF.groupBy("movieId").agg(avg("rating").alias("avgRating"))
// 将电影名称与平均评分进行关联
val movieRatingsDF = moviesDF.join(avgRatingsDF, Seq("movieId"))
// 按平均评分从高到低排序，并取前10部电影
val top10MoviesDF = movieRatingsDF.orderBy(desc("avgRating")).limit(10)

// 显示结果
top10MoviesDF.show()

这段代码首先计算每部电影的平均评分，然后将电影名称与平均评分进行关联，最后按平均评分从高到低排序，并取前10部电影。在这个示例中，我们使用了Spark SQL的函数功能，如avg、desc等。