为了帮助你解决《综合实训报告模板》中的三个问题，我将提供详细的PyCharm代码示例及其解释。以下是针对每个任务的具体实现： ### 1. 使用Spark RDD编程读取本地文件并统计平均评分最高的前20部电影信息 #### 代码示例 ```python from pyspark import SparkConf, SparkContext # 初始化Spark配置 conf = SparkConf().setAppName("MovieRatings").setMaster("local") sc = SparkContext(conf=conf) # 读取movies.csv文件 movies = sc.textFile("path/to/movies.csv") header = movies.first() movies = movies.filter(lambda line: line != header).map(lambda line: line.split(",")) # 读取ratings.csv文件 ratings = sc.textFile("path/to/ratings.csv") header = ratings.first() ratings = ratings.filter(lambda line: line != header).map(lambda line: (line.split(",")[1], float(line.split(",")[2]))) # 计算每部电影的平均评分 movie_ratings = ratings.groupByKey().mapValues(lambda x: sum(x) / len(x)) top_movies = movie_ratings.join(movies).map(lambda x: (x[1][1], x[1][0])).sortBy(lambda x: x[1], ascending=False).take(20) # 输出结果 for movie in top_movies: print(f"Movie ID: {movie[0]}, Movie Title: {movie[1][0]}, Average Rating: {movie[1][1]}") # 停止SparkContext sc.stop() ``` #### 代码解释 1. **初始化Spark配置**：设置应用程序名称和运行模式。 2. **读取`movies.csv`文件**：跳过标题行，将每行数据按逗号分割成列表。 3. **读取`ratings.csv`文件**：跳过标题行，提取电影ID和评分，并将其转换为键值对形式。 4. **计算每部电影的平均评分**：使用`groupByKey`将所有评分归类到同一电影ID下，然后计算平均值。 5. **合并电影信息和评分**：将电影ID和平均评分与电影名称合并，并按平均评分降序排序。 6. **输出结果**：打印平均评分最高的前20部电影的信息。 7. **停止SparkContext**：释放资源。 ### 2. 使用Spark SQL编程读取本地文件并完成数据分析和统计任务 #### 代码示例 ```python from pyspark.sql import SparkSession # 初始化SparkSession spark = SparkSession.builder.appName("MovieAnalysis").getOrCreate() # 读取movies.csv文件 movies_df = spark.read.csv("path/to/movies.csv", header=True, inferSchema=True) movies_df.createOrReplaceTempView("movies") # 读取ratings.csv文件 ratings_df = spark.read.csv("path/to/ratings.csv", header=True, inferSchema=True) ratings_df.createOrReplaceTempView("ratings") # 任务1：查找电影评分个数超过5000且平均评分最高的前10部电影 query1 = """ SELECT m.movieId, m.title, AVG(r.rating) AS avg_rating FROM movies m JOIN ratings r ON m.movieId = r.movieId GROUP BY m.movieId, m.title HAVING COUNT(r.rating) > 5000 ORDER BY avg_rating DESC LIMIT 10 """ result1 = spark.sql(query1) result1.show() result1.write.jdbc(url="jdbc:mysql://localhost:3306/mydb", table="top_rated_movies", mode="overwrite", properties={"user": "root", "password": "password"}) # 任务2：按照genres字段分类统计每类电影的平均评分 query2 = """ SELECT explode(split(m.genres, '\\|')) AS genre, AVG(r.rating) AS avg_rating FROM movies m JOIN ratings r ON m.movieId = r.movieId GROUP BY genre """ result2 = spark.sql(query2) result2.show() result2.write.jdbc(url="jdbc:mysql://localhost:3306/mydb", table="genre_avg_ratings", mode="overwrite", properties={"user": "root", "password": "password"}) # 任务3：统计评分次数最多的前10部电影 query3 = """ SELECT m.movieId, m.title, COUNT(r.rating) AS rating_count FROM movies m JOIN ratings r ON m.movieId = r.movieId GROUP BY m.movieId, m.title ORDER BY rating_count DESC LIMIT 10 """ result3 = spark.sql(query3) result3.show() result3.write.jdbc(url="jdbc:mysql://localhost:3306/mydb", table="most_rated_movies", mode="overwrite", properties={"user": "root", "password": "password"}) # 任务4：统计每个用户的评分次数，找出评分次数最多的用户 query4 = """ SELECT userId, COUNT(rating) AS rating_count FROM ratings GROUP BY userId ORDER BY rating_count DESC LIMIT 1 """ result4 = spark.sql(query4) result4.show() # 任务5：找出评分时间最早和最晚的电影及其对应评分时间 query5 = """ SELECT m.movieId, m.title, MIN(r.timestamp) AS min_timestamp, MAX(r.timestamp) AS max_timestamp FROM movies m JOIN ratings r ON m.movieId = r.movieId GROUP BY m.movieId, m.title """ result5 = spark.sql(query5) result5 = result5.withColumn("min_timestamp", F.to_timestamp(F.col("min_timestamp"))) result5 = result5.withColumn("max_timestamp", F.to_timestamp(F.col("max_timestamp"))) result5.show() # 停止SparkSession spark.stop() ``` #### 代码解释 1. **初始化SparkSession**：创建Spark会话以使用Spark SQL功能。 2. **读取CSV文件**：使用`read.csv`方法读取`movies.csv`和`ratings.csv`文件，并创建临时视图以便后续查询。 3. **任务1**：查找电影评分个数超过5000且平均评分最高的前10部电影。执行SQL查询并将结果写入MySQL数据库。 4. **任务2**：按照genres字段分类统计每类电影的平均评分。执行SQL查询并将结果写入MySQL数据库。 5. **任务3**：统计评分次数最多的前10部电影。执行SQL查询并将结果写入MySQL数据库。 6. **任务4**：统计每个用户的评分次数，找出评分次数最多的用户。执行SQL查询并显示结果。 7. **任务5**：找出评分时间最早和最晚的电影及其对应评分时间。执行SQL查询并格式化时间戳。 8. **停止SparkSession**：释放资源。 ### 3. 使用Spark Streaming读取套接字流并实时统计平均评分最高的前3部电影信息 #### 代码示例 ```python from pyspark.streaming import StreamingContext from pyspark.sql import SparkSession import socket # 初始化SparkSession spark = SparkSession.builder.appName("StreamingMovieRatings").getOrCreate() # 创建StreamingContext ssc = StreamingContext(spark.sparkContext, batchDuration=10) # 创建DStream从套接字接收数据 lines = ssc.socketTextStream("localhost", 9999) # 解析每行数据 def parse_line(line): parts = line.split("|") return (parts[0].strip(), float(parts[1].strip())) parsed_lines = lines.map(parse_line) # 计算每部电影的平均评分 def update_total_count(new_values, current_sum): if current_sum is None: current_sum = (0, 0) for value in new_values: current_sum = (current_sum[0] + value, current_sum[1] + 1) return current_sum movie_ratings = parsed_lines.updateStateByKey(update_total_count) # 计算平均评分并找到评分最高的前3部电影 def compute_top_3_movies(time, rdd): top_3_movies = rdd.mapValues(lambda x: x[0] / x[1]).transform(lambda rdd: rdd.sortBy(lambda x: -x[1])).take(3) print(f"Top 3 movies at time {time}:") for movie in top_3_movies: print(f"Movie Name: {movie[0]}, Average Rating: {movie[1]}") movie_ratings.foreachRDD(compute_top_3_movies) # 启动流处理 ssc.start() ssc.awaitTermination() ``` #### 代码解释 1. **初始化SparkSession**：创建Spark会话以使用Spark SQL功能。 2. **创建StreamingContext**：设置批处理间隔时间为10秒。 3. **创建DStream**：从套接字接收数据，地址为`localhost`，端口为`9999`。 4. **解析每行数据**：定义函数`parse_line`将每行数据解析为`(电影名称, 评分)`的元组。 5. **更新状态**：定义函数`update_total_count`用于更新每部电影的总评分和评分次数。 6. **计算平均评分**：定义函数`compute_top_3_movies`计算每部电影的平均评分，并找到评分最高的前3部电影。 7. **启动流处理**：启动流处理并等待终止。 希望这些代码示例和解释对你有所帮助！如果有任何问题或需要进一步的帮助，请随时告诉我。