别再死记硬背了！用Spark实战电影评分分析，手把手教你搞定Join操作与数据清洗
用Spark实战电影评分分析：从数据清洗到Join操作的完整指南
每次看到电影评分网站上的"Top 100"榜单，你有没有好奇过背后的数据处理逻辑？作为Spark初学者，你可能已经啃了不少官方文档，但面对真实数据集时依然无从下手。本文将带你用MovieLens数据集，从零构建一个电影评分分析系统，在实战中掌握Join操作与数据清洗的核心技巧。
1. 环境准备与数据理解
在开始编码前，我们需要搭建好开发环境并理解数据集结构。推荐使用Spark 3.x版本，它提供了更优的SQL优化器和性能提升。数据集方面，MovieLens 100K是个不错的起点，包含：
- 评分数据（ratings.dat）：用户ID、电影ID、评分（1-5分）、时间戳
- 电影数据（movies.dat）：电影ID、标题、类型
提示：可以从GroupLens官网下载数据集，解压后你会看到.dat文件，这是用双冒号(::)分隔的文本格式。
安装依赖只需一行命令：
pip install pyspark==3.3.1 pandas初始化SparkSession时，建议开启这些配置提升性能：
from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("MovieRatingAnalysis") \ .config("spark.sql.shuffle.partitions", "8") \ .config("spark.executor.memory", "2g") \ .getOrCreate()2. 数据加载与初步清洗
原始数据往往存在各种问题，我们需要先进行基础清洗。用Spark读取.dat文件时，指定分隔符和列名：
ratings = spark.read.csv( "ratings.dat", sep="::", schema="user_id int, movie_id int, rating double, timestamp long" ) movies = spark.read.csv( "movies.dat", sep="::", schema="movie_id int, title string, genres string" )常见的数据质量问题包括：
- 缺失值：某些评分为null
- 异常值：评分不在1-5范围内
- 重复记录：同一用户对同一电影多次评分
清洗操作示例：
clean_ratings = ratings.filter( (ratings.rating.isNotNull()) & (ratings.rating >= 1) & (ratings.rating <= 5) ).dropDuplicates(["user_id", "movie_id"])3. 核心分析：计算电影平均分
这才是真正体现Spark价值的地方。我们需要：
- 按电影ID分组
- 计算每个电影的平均分
- 过滤出高评分电影（如>4.0）
用DataFrame API实现：
from pyspark.sql import functions as F avg_ratings = clean_ratings.groupBy("movie_id") \ .agg(F.avg("rating").alias("avg_rating")) \ .filter("avg_rating > 4.0")如果数据量很大，可以优化计算：
# 使用approxQuantile快速识别异常值 bounds = clean_ratings.approxQuantile("rating", [0.25, 0.75], 0.05) iqr = bounds[1] - bounds[0] valid_range = [bounds[0] - 1.5*iqr, bounds[1] + 1.5*iqr]4. Join操作：关联电影信息
现在有了高分电影ID，但用户想看的是电影名称而非ID。这正是Join的用武之地：
top_movies = avg_ratings.join( movies, avg_ratings.movie_id == movies.movie_id, "inner" ).select("title", "avg_rating")Spark支持多种Join类型，选择策略很重要：
| Join类型 | 适用场景 | 性能影响 |
|---|---|---|
| inner | 只保留两边都匹配的记录 | 最快 |
| left | 保留左表所有记录 | 中等 |
| right | 保留右表所有记录 | 中等 |
| full | 保留所有记录 | 最慢 |
注意：大数据集Join可能引发shuffle，合理设置
spark.sql.shuffle.partitions很关键
5. 性能优化与调试技巧
当处理GB级数据时，这些技巧能帮你节省数小时：
缓存策略选择：
movies.cache() # 会被多次使用的表 avg_ratings.persist(StorageLevel.MEMORY_AND_DISK) # 内存不足时溢写到磁盘执行计划分析：
top_movies.explain()广播小表优化：
from pyspark.sql.functions import broadcast top_movies = avg_ratings.join( broadcast(movies), # 小于10MB的表适合广播 "movie_id" )常见问题排查：
- OOM错误：增加executor内存或减少分区数
- 数据倾斜：使用
salt技术分散热点key - Join耗时：检查是否该使用广播Join
6. 结果展示与业务解读
最终我们可以将结果导出为CSV，或用可视化工具展示：
top_movies.orderBy("avg_rating", ascending=False) \ .limit(100) \ .write.csv("top_movies.csv")业务角度的一些发现：
- 高评分电影多集中在特定类型（如纪录片）
- 经典老片评分普遍高于新片
- 某些导演的多部作品上榜
这些洞察可以帮助：
- 推荐系统优化
- 库存采购决策
- 用户画像完善
7. 扩展思路：从分析到生产
如果想将这套流程产品化，还需要：
- 自动化：用Airflow调度每日跑批
- 监控：跟踪评分分布变化
- 实时化：改用Structured Streaming处理新评分
# 流式处理示例 streaming_ratings = spark.readStream \ .schema(ratings.schema) \ .csv("hdfs://new_ratings/")处理真实业务数据时，你可能会遇到：
- 电影更名导致ID冲突
- 用户刷分行为
- 类型标签不准确
这些都需要在ETL流程中加入额外处理逻辑。