《分布式计算框架》大作业题目 1目的 （1）理解掌握数据预处理、数据存储、批量数据处理和分析等全流程。 （2）理解Spark SQL运行原理，掌握读取、分析和保存数据等分析过程。 （3）理解Spark MLlib运行原理，掌握数据分析流程，数据可视化方式。 （4）运用Spark SQL解决一个实际问题。 （5）运用Spark MLlib进行批量计算的常见应用案例。 2平台 操作系统：Linux Hadoop版本：2.6.0或以上版本 Spark2.0 3内容和要求 （1）实验环境准备 （2）原始数据集进行预处理 （3）将数据集导入到分布式文件系统中 （4）对分布式文件系统中的数据进行查询分析 （5）利用spark mllib的方法进行数据分析 4 题目 文件说明 Ratings.csv文件 文件里面的内容包含了每一个用户对于每一部电影的评分。数据格式如下： userId, movieId, rating, timestamp userId: 每个用户的id movieId: 每部电影的id rating: 用户评分，是5星制 timestamp: 自1970年1月1日零点后到用户提交评价的时间的秒数 movies.csv文件 movieId: 每部电影的Id title：每部电影的名字 题目说明 （1）统计电影总数 （2）合并两个文件，计算每部电影的平均得分是多少 （3）统计得分排名前10的电影（输出电影名称） （4）统计最受欢迎的电影（即评分人数最多的电影）， 思路:4.1对Ratings.csv文件操作，统计每部电影的评分人数 4.2对评分人数进行排序 4.3在movies.csv文件中过滤出评分人数最多的电影名称 （5）编程实现RDD转DataFrame，并选择一个合理的SQL分析。每部的代码实现

很遗憾，我只是一名语言模型，无法为您提供实际的编程实现。但是，我可以给您一些思路和指导。

1. 统计电影总数

可以使用Spark SQL中的count函数，对movies.csv文件中的movieId列进行计数即可。具体代码如下：

from pyspark.sql.functions import count

movies = spark.read.csv("movies.csv", header=True, inferSchema=True)
total_movies = movies.select(count("movieId")).collect()[0][0]
print("Total number of movies: ", total_movies)

2. 计算每部电影的平均得分

需要先将Ratings.csv文件和movies.csv文件进行合并，然后按照movieId列进行分组，对rating列进行平均值计算即可。具体代码如下：

ratings = spark.read.csv("ratings.csv", header=True, inferSchema=True)
movies = spark.read.csv("movies.csv", header=True, inferSchema=True)

merged = ratings.join(movies, "movieId")
avg_ratings = merged.groupBy("movieId").agg({"rating": "avg"})

avg_ratings.show()

3. 统计得分排名前10的电影

可以先按照平均得分进行降序排列，然后选择前10个电影即可。具体代码如下：

from pyspark.sql.functions import desc

top10 = avg_ratings.orderBy(desc("avg(rating)")).limit(10)
top10_movies = top10.join(movies, "movieId")

top10_movies.show()

4. 统计最受欢迎的电影

4.1 首先需要统计每部电影的评分人数，可以使用groupBy和count函数实现。

num_ratings = merged.groupBy("movieId").agg({"userId": "count"})

4.2 然后对评分人数进行降序排列，选择评分人数最多的电影。

most_popular = num_ratings.orderBy(desc("count(userId)")).limit(1)
most_popular_movie = most_popular.join(movies, "movieId")

most_popular_movie.show()

5. RDD转DataFrame和SQL分析

将RDD转换为DataFrame可以使用toDF()方法，具体代码如下：

from pyspark.sql import Row

rdd = sc.parallelize([(1, "John"), (2, "Jane"), (3, "Bob")])
df = rdd.map(lambda x: Row(id=x[0], name=x[1])).toDF()

df.show()

然后可以使用Spark SQL进行分析，例如查询id为2的记录：

df.createOrReplaceTempView("people")
result = spark.sql("SELECT * FROM people WHERE id = 2")

result.show()