别再死记硬背了！用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 pyspark3.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::, schemauser_id int, movie_id int, rating double, timestamp long ) movies spark.read.csv( movies.dat, sep::, schemamovie_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技术分散热点keyJoin耗时检查是否该使用广播Join6. 结果展示与业务解读最终我们可以将结果导出为CSV或用可视化工具展示top_movies.orderBy(avg_rating, ascendingFalse) \ .limit(100) \ .write.csv(top_movies.csv)业务角度的一些发现高评分电影多集中在特定类型如纪录片经典老片评分普遍高于新片某些导演的多部作品上榜这些洞察可以帮助推荐系统优化库存采购决策用户画像完善7. 扩展思路从分析到生产如果想将这套流程产品化还需要自动化用Airflow调度每日跑批监控跟踪评分分布变化实时化改用Structured Streaming处理新评分# 流式处理示例 streaming_ratings spark.readStream \ .schema(ratings.schema) \ .csv(hdfs://new_ratings/)处理真实业务数据时你可能会遇到电影更名导致ID冲突用户刷分行为类型标签不准确这些都需要在ETL流程中加入额外处理逻辑。