简介基于Spark平台对TMDB电影数据进行完整分析的毕业设计项目包内含源代码与说明文档主要面向计算机相关专业的在校学生、数据分析入门者以及需要课程设计或毕业设计参考的开发者可用于大数据课程作业、项目初期立项演示或毕业设计答辩等场景。项目以Scala编写Spark分析主程序围绕电影评分、类型、上映年份等多个维度进行数据清洗、统计聚合与结果分析并通过ECharts生成交互式可视化页面帮助使用者清晰理解Spark SQL和DataFrame的完整处理流程。资源共包含45个文件压缩包仅2.55MB涵盖17个HTML结果页面、15个JSON数据文件、6个JS交互脚本、2个Python辅助脚本、1个Scala主程序以及README说明文档其中HTML负责展示分析结果JSON提供数据支撑JS完成前端交互Python与Scala处理后台逻辑目录结构清晰便于按模块阅读和二次开发。项目代码经过运行测试功能均可正常使用作者答辩评审平均分达96分如遇运行问题还可私聊联系作者获取远程教学指导。目前已有244人学习下载适合作为业务型数据分析实战练习也适合在现有代码基础上进行功能扩展作为毕业设计或课程设计的可靠参考方案。1. TMDB 电影数据集用 Spark 做分析的几个真实理由TMDB 电影数据集常被当成 Spark 入门的练习题但它并不是一张干净的表。movies_metadata.csv、credits.csv、ratings_small.csv 三个文件凑到一起字段里混着 JSON、空字符串和不规范的日期单靠 pandas 也能处理但一旦分析口径要反复调整DataFrame 的惰性求值和 SQL 表达能力能把这类探索工作拉回结构化流程。这也是“基于 Spark 平台 TMDB 电影数据分析”这个标题真正要回答的问题环境参数怎么备、字段怎么清、聚合怎么做、结果怎么验证。下面五章按一条能本地跑通的分析主线展开从 SparkSession 配置到 Schema 声明再到数据清洗、票房聚合、窗口排名和输出验证每一段都可以直接抄进自己的 Spark 项目里运行。动手的人最好已经搭好一个小规模的 Spark Standalone 或 YARN 客户端开发环境剩下的靠代码说话。2. Spark 环境参数与 Schema 设计读取 TMDB CSV 前的准备数据处理的第一步不是写聚合而是决定读取方式和字段类型。这一章解决两件事SparkSession 的参数怎么给TMDB 原始文件的 Schema 怎么声明。两个问题在几万行数据上容易一带而过等作业交付后才发现结果在类型转换环节已经悄悄错了。2.1 读取 TMDB 三个数据文件前要确定分析边界“电影数据分析”落在哪个粒度会影响后面所有代码。我一般会把第一版范围定为“电影元数据 用户评分”两个文件credits.csv 留作扩展。理由很直接片名、预算、票房、类型、日期、评分都集中在 movies_metadata.csv 一个文件里分析性价比最高credits 一旦要 join 角色、导演关系依赖链会被拉长第一次跑通主线时没必要引入。文件包含内容本次分析中的角色movies_metadata.csv片名、预算、票房、类型、日期、评分主表ratings_small.csv用户对电影的评分记录评分分布辅助credits.csv演员、导演、职位信息后续扩展不参与第一版确定边界之后再决定读取路径。本地开发用file:///home/xxx/tmdb-spark/绝对路径相对路径在 Jupyter 和 spark-submit 之间经常不一致。如果提交到 YARN 模式数据文件要提前放到 HDFS并把读取路径改成hdfs:///data/tmdb/movies_metadata.csv否则 driver 本地能找到文件而 executor 全部报 file not found这是集群模式下最常见的启动失败原因。2.2 手动 Schema 声明代替类型推断的三个理由Spark 的read.csv不指定 schema 时会自动推断对 TMDB 文件这个机制会带来三种典型错误。第一id 列在部分行里有非数字字符推断结果可能变成 DoubleType后续与评分表的整数 movieId join 时永远匹配不上。第二budget 和 revenue 里混着 0、空字符串和上亿的大数推断出来的字段类型要么不够长要么把整列定为 string聚合时还得再转。第三genres 字段是带转义引号的 JSON类型推断只负责最外层的 StringType实际解析仍然要手动声明目标类型。手动声明 schema 后CSV 解析器遇到类型不一致的行时能按既定策略处理而不是自动切换类型。先定义主表结构from pyspark.sql.types import ( StructType, StructField, StringType, LongType, DoubleType, IntegerType ) metadata_schema StructType([ StructField(budget, LongType(), True), StructField(genres, StringType(), True), StructField(homepage, StringType(), True), StructField(id, StringType(), True), StructField(keywords, StringType(), True), StructField(original_language, StringType(), True), StructField(original_title, StringType(), True), StructField(overview, StringType(), True), StructField(popularity, DoubleType(), True), StructField(release_date, StringType(), True), StructField(revenue, LongType(), True), StructField(runtime, IntegerType(), True), StructField(title, StringType(), True), StructField(vote_average, DoubleType(), True), StructField(vote_count, IntegerType(), True) ])声明里特意把 id 保留成 StringType留到清洗阶段统一转数值budget 和 revenue 用 LongType 而不是 IntegerType因为单部电影票房超过 21 亿时 IntegerType 会溢出。release_date 也保留字符串源数据里有年份、斜杠日期和空值三种形态直接定成 DateType 会有一批行变成 null。读入文件metadata_raw ( spark.read .option(header, True) .option(quote, \) .option(escape, \) .option(mode, PERMISSIVE) .schema(metadata_schema) .csv(data/tmdb/movies_metadata.csv) ) print(metadata_raw.count()) metadata_raw.printSchema()这几个 option 作用不同。header 把首行作为列名quote 指定字段包裹符escape 指定字段内的转义符缺少它会直接截断 genres 这类 JSONmode 设为 PERMISSIVE 表示遇到溢出不匹配时用 null 填充而不是让整个作业失败schema 与 mode 配合决定哪些行保留。第一次运行时先 count 和 printSchema确认读入行数没有明显少于源文件行数再往下走。提示如果 count 结果明显小于文件行数优先去查文件里是否有长字段跨行。genres、overview 这类字段被双引号包着时内部若含换行就会出现列错位手动 Schema 也救不回来需要加上.option(multiline, True)重新读取。2.3 SparkSession 的几个运行参数怎么设SparkSession 的配置项不用多针对 TMDB 数据规模调三个就够from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(tmdb-movie-analysis) \ .config(spark.sql.shuffle.partitions, 12) \ .config(spark.sql.autoBroadcastJoinThreshold, 104857600) \ .config(spark.sql.session.timeZone, UTC) \ .getOrCreate()参数说明spark.sql.shuffle.partitions控制 groupBy 和 join 后 shuffle 的分区数。默认值是 200在单机跑 TMDB 这种几万行数据会出现大量空 task设成 12 表示聚合后最多 12 个分区并发计算降低调度开销。spark.sql.autoBroadcastJoinThreshold默认 10MB设成 104857600 字节即 100MB 后连接评分表时有机会自动走 BroadcastHashJoin 而不是 SortMergeJoin。spark.sql.session.timeZone指定会话时区提交节点和集群节点时区不一致时to_date 和 year 的结果可能差一天显式声明可以避免这种偏移。这几个参数里最容易踩坑的是 shuffle.partitions 调太大。在一个只有 4 核的环境设置 200Spark UI 里会出现大量空 task日志看起来是 Failed 但作业仍然成功实际是重试空任务导致的假失败现象。调参之后用同一个聚合查询对比执行时间比凭感觉改配置更可靠。3. DataFrame 清洗 TMDB 原始数据JSON 解析、空值和日期转换读取只是开始TMDB 原始文件里有两类不干净数值列的 0 和 null 混在一起以及 genres 字段里嵌着 JSON 需要按结构提取。这一章处理从原始行到可分析表的整段转换也是标题里“源代码”价值最集中的位置。3.1 字符串 id 转 Long 并清除无法识别行第 2 章把 id 保留成字符串这一节先转数值再过滤from pyspark.sql import functions as F metadata metadata_raw \ .withColumn( movie_id, F.when(F.col(id).rlike(^[0-9]$), F.col(id).cast(long)) .otherwise(F.lit(None)) ) \ .filter(F.col(movie_id).isNotNull()) \ .drop(id) metadata.select(title, movie_id).show(5, truncateFalse) print(有效行数:, metadata.count())逻辑说明rlike(^[0-9]$)是正则匹配只保留纯数字的 id。匹配成功的行被 cast 成 long失败的行补 null随后 filter 丢弃。先正则后转换是为了避免 Spark 对非数字字符串直接 cast 时抛出异常。drop 删除旧的字符串列后续所有 join 统一用 movie_id 作为连接键。执行后如果有效行数比 CSV 原始行数少很多先看被过滤掉的行长什么样用metadata_raw.filter(~F.col(id).rlike(^[0-9]$)).select(id, title).show(10, False)输出样本。多数情况是源数据里的脏行去掉不影响总体分析如果占比超过 5%要回去确认 CSV 文件是不是下载不完整。3.2 预算、票房与利润字段的零值处理budget 和 revenue 为 0 的电影在 TMDB 数据里占相当比例这些不是真实的 0而是公开资料缺失。直接用 0 参与平均票房计算会把均值显著拉低所以分析时按两套口径处理一套保留全量用于统计影片数量另一套过滤营收场景专门计算商业指标。movies_numeric metadata \ .withColumn(budget_num, F.coalesce(F.col(budget), F.lit(0))) \ .withColumn(revenue_num, F.coalesce(F.col(revenue), F.lit(0))) \ .withColumn(profit, F.col(revenue_num) - F.col(budget_num)) movies_business movies_numeric.filter(F.col(revenue_num) 0) \ .filter(F.col(budget_num) 0) print(全量电影:, movies_numeric.count()) print(含预算与票房数据的电影:, movies_business.count())coalesce 从参数里取第一个非 null 值这里把 null 统一补成 0。profit 可能是负数代表亏损。过滤条件拆成两个 filter比直接写 and 表达式更容易维护和替换。打印两个 count 能直观看到还有多少样本可用于商业分析如果只剩几百行就不要用平均值指标改成中位数。注意不要覆盖原来的 budget 和 revenue 列。原始值后面反查数据或调整口径时还要用新增派生列比原地覆盖安全得多。3.3 用 from_json 解析 genres 和 keywords 字段genres 字段形如[{id: 28, name: Action}, {id: 12, name: Adventure}]。要用 from_json 把它解析成结构体数组才能进一步展开from pyspark.sql.types import ArrayType, StructType, StructField, IntegerType genre_schema ArrayType( StructType([ StructField(id, IntegerType(), True), StructField(name, StringType(), True) ]) ) movies_mapped movies_business \ .withColumn( genres_arr, F.from_json(F.col(genres), genre_schema) ) \ .withColumn( genre_names, F.when(F.col(genres_arr).isNotNull(), F.expr(transform(genres_arr, g - g.name))) .otherwise(F.array().cast(arraystring)) ) \ .withColumn(primary_genre, F.element_at(F.col(genre_names), 1)) movies_mapped.select(title, genre_names, primary_genre).show(5, truncateFalse)这三个派生列的作用不同。genres_arr 是结构体数组每个元素有 id 和 namegenre_names 通过 transform 高阶函数取出 name 转换成纯字符串数组为后面的 explode 做准备primary_genre 取数组第一个元素作为主类型。from_json 的两个参数一个是列名一个是目标类型。如果 JSON 字段名或嵌套层级与 schema 不匹配解析结果是整体 null 而不是部分 null。写这段时必须避开一个常见坑目标类型声明成ArrayType(StructType(...))如果误写成单纯的 StructType解析结果对不上数组结构整列全是 null。keywords 字段结构与 genres 完全相同复制同一套声明即可。这里还有个性能细节from_json 代价不低如果清洗时 genres 和 keywords 都要解析别写两遍 withColumn而是用 select 一次性扩展出多列减少对同一条原始 JSON 的重复反序列化。3.4 日期标准化与非法行过滤release_date 中混有yyyy-MM-dd和yyyy/MM/dd两种格式对年份聚合的影响很大movies_final movies_mapped \ .filter(F.col(release_date).isNotNull()) \ .withColumn( release_dt, F.to_date(F.col(release_date), yyyy-MM-dd) ) \ .filter(F.col(release_dt).isNotNull()) \ .withColumn(release_year, F.year(F.col(release_dt)))逻辑说明to_date 按yyyy-MM-dd解析遇到斜杠日期返回 null第二个 filter 把这些行排除保证“有上映年份”的分析建立在真实可解析的日期上。如果业务上必须救回斜杠日期要用 unix_timestamp 分次试探两种格式再取非 null 值不能只改 to_date 的格式参数。过滤时也可以把空日期行保留成一个单独的“unknown”分组但 TMDB 数据里这类行极少直接过滤掉对结论几乎没有影响。4. 票房聚合与窗口排名TMDB 电影数据的多维统计实战清洗后的 movies_final 可以复现三类最常见的分析年度票房走势、类型收益差异、每个类型下的 Top-N 电影。这三个场景分别对应 groupBy 聚合、explode 展开、窗口函数排名是 Spark 数据分析案例中出场频率最高的操作组合。4.1 年度票房聚合与平均分统计口径先按年汇总看数据质量和趋势走向yearly_stats movies_final \ .groupBy(release_year) \ .agg( F.count(movie_id).alias(movie_count), F.sum(revenue_num).alias(total_revenue), F.avg(vote_average).alias(avg_rating), F.round(F.avg(profit), 2).alias(avg_profit) ) \ .orderBy(F.desc(release_year)) yearly_stats.show(10, truncateFalse)这条链里最容易忽视的是 avg_rating 会把 vote_count 为 0 的电影也并进平均。无人评分的电影 vote_average 通常为 0直接把平均值拉得很低。更严谨的口径是先过滤出有足够投票数的电影yearly_rated movies_final \ .filter(F.col(vote_count) 20) \ .groupBy(release_year) \ .agg( F.count(movie_id).alias(rated_movie_count), F.round(F.avg(vote_average), 2).alias(avg_rating), F.percentile_approx(F.col(vote_average), 0.5).alias(median_rating) ) \ .orderBy(F.desc(release_year)) yearly_rated.show(10, truncateFalse)percentile_approx 是近似分位数计算第二个参数 0.5 表示中位数。相比精确百分位在大数据量下快得多在 TMDB 这个规模结果与精确计算基本一致。这里同时保留 avg_rating 和 median_rating是因为电影评分分布是长尾的一部高分小众电影会把平均分拉高而中位数更能反映大众观感。如果某一年 movie_count 和 total_revenue 同时明显下降先检查 release_year 是否大量为 null而不是怀疑聚合语法。4.2 用 explode 展开类型字段做多类型统计类型维度分析需要把一部电影的多类型拆成多行genre_exploded movies_final \ .select( movie_id, title, budget_num, revenue_num, profit, vote_average, genre_names ) \ .withColumn(genre_name, F.explode(F.col(genre_names))) genre_stats genre_exploded \ .groupBy(genre_name) \ .agg( F.count(movie_id).alias(movie_cnt), F.avg(revenue_num).alias(avg_revenue), F.avg(profit).alias(avg_profit), F.avg(vote_average).alias(avg_rating) ) \ .orderBy(F.desc(movie_cnt)) genre_stats.show(20, truncateFalse)explode 把数组的每个元素生成一行因此一部同时属于 Action 和 Adventure 的电影会出现在两个类型分组里。这时的 movie_cnt 语义是“该类型下电影出现次数”不是去重后的电影数量。要表达“有多少部电影属于该类型”先用 movie_id 和 genre_name 做 distinct 再 count。select 里只保留后续需要的列可以减少 explode 之后管道中传输的数据量这个习惯在数据量大时能明显降低 shuffle 成本。看结果时注意 avg_revenue 会被少数商业大片拉高类型间的真实差异建议再看一版中位数。genre_exploded 已经是公共中间结果后面所有类型维度的二次聚合都复用它不再重复 explode。4.3 用窗口函数在类型内部算 Top-N 排名常见的需求是每个类型下票房最高的 10 部电影。窗口函数解决“组内排序后取前 N 条”比 groupBy 之后手工 collect_list 再截断直观很多from pyspark.sql import Window genre_win Window.partitionBy(genre_name).orderBy(F.desc(revenue_num)) top_movies genre_exploded \ .withColumn(rank, F.row_number().over(genre_win)) \ .filter(F.col(rank) 10) \ .select(genre_name, title, revenue_num, vote_average, rank) \ .orderBy(genre_name, rank) top_movies.show(20, truncateFalse)执行逻辑是partitionBy 按 genre_name 分组orderBy 在组内按票房降序row_number 生成从 1 开始的连续编号最终 filter 保留前 10 条。这个写法不会丢其他列也不需要二次 join 回主表。排名函数的选择在票房并列时差别很大。row_number 遇到相同票房会随机给出不同名次rank 和 dense_rank 都会让并列电影拿到相同名次区别在于 rank 会在并列后跳过后续编号。业务上做票房榜单时通常要求并列呈现应优先用 rank再按 title 排序保证输出稳定。5. 缓存、广播连接与结果验证把 Spark 作业收尾的调优手段分析代码跑通之后收尾工作集中在三件容易被跳过的事缓存中间表、用广播替代 shuffle、把关键结果落盘并验证。5.1 缓存选择标准什么时候 cache()什么时候别用对 TMDB 这个规模元数据只有几万行很多时候不需要缓存缓存反而增加 executor 内存压力。判断标准是同一个 DataFrame 是否要被后续两个以上的任务复用。上面生成的 genre_exploded 就是典型它同时服务于类型统计和 Top-N 排名from pyspark import StorageLevel genre_exploded.persist(StorageLevel.MEMORY_AND_DISK) genre_exploded.count()persist 用于指定存储级别MEMORY_AND_DISK 表示内存放不下时落到磁盘。cache() 默认是 MEMORY_ONLY内存不够就重新计算对反复使用的中间结果MEMORY_AND_DISK 更符合本地开发场景。缓存后的第一次 count 才触发真实计算后续查询直接读缓存。源数据逻辑一旦变更记得先 unpersist 再重新 persist不要拿旧缓存当新结果。5.2 用 broadcast 连接评分数据减少 shuffle 成本rating 表只有约十万行非常小适合广播连接。上一章末尾把 autoBroadcastJoinThreshold 调成 100MB这里在代码里显式标记ratings_raw ( spark.read .option(header, True) .csv(data/tmdb/ratings_small.csv) .select( F.col(movieId).cast(long).alias(movie_id), F.col(rating).cast(double).alias(rating) ) .filter(F.col(movie_id).isNotNull()) ) ratings_joined movies_final.join( F.broadcast(ratings_raw), movie_id, inner ) print(ratings_joined.count())F.broadcast 是显式广播标记Spark 会把 ratings_raw 序列化分发到每个 executor对 movies_final 逐条匹配。这样 shuffle 只发生在 movies_final 一侧评分表不参与全量排序。反过来如果小表很大比如超过 100MB 还强制广播会造成 driver 端 GC 压力和 executor 内存溢出不要滥用。可以在 Spark UI 的 SQL 标签页确认 join 节点是不是 BroadcastHashJoin如果不是去检查阈值配置是否生效。5.3 落盘与数值正确性验证最后把统计结果输出成 parquet同时做一次数值校验避免上游清洗逻辑悄悄带歪结果movies_final.write.mode(overwrite).parquet(output/movies_clean) yearly_rated.write.mode(overwrite).parquet(output/yearly_stats) top_movies.write.mode(overwrite).parquet(output/top_movies) check movies_final.agg( F.count(movie_id).alias(total_movies), F.sum(revenue_num).alias(total_revenue), F.avg(vote_average).alias(avg_all_rating) ).collect() assert check[0][0] 3000, 有效电影数过少检查过滤条件 assert check[0][1] 0, 总票房非正检查收入字段类型 print(check)把断言写进脚本而不是只在控制台看一眼是为了让流程可以放进定时任务或 CI。断言阈值要按实际数据定义这里 3000 是 TMDB 全量数据清洗后的合理下限。panel 更精细的验证方式是把聚合结果和原始 CSV 的唯一 id 数对比如果两边数字差太多多半是第 3 章的 id 过滤或日期过滤过于严格。落盘完成后用spark.read.parquet(output/movies_clean).createOrReplaceTempView(movies)注册成临时视图后续任何聚合结果都可以用一行 SQL 快速反查这是调整分析参数时最省事的验证手段。本文还有配套的精品资源点击获取
简介基于Spark平台对TMDB电影数据进行完整分析的毕业设计项目包内含源代码与说明文档主要面向计算机相关专业的在校学生、数据分析入门者以及需要课程设计或毕业设计参考的开发者可用于大数据课程作业、项目初期立项演示或毕业设计答辩等场景。项目以Scala编写Spark分析主程序围绕电影评分、类型、上映年份等多个维度进行数据清洗、统计聚合与结果分析并通过ECharts生成交互式可视化页面帮助使用者清晰理解Spark SQL和DataFrame的完整处理流程。资源共包含45个文件压缩包仅2.55MB涵盖17个HTML结果页面、15个JSON数据文件、6个JS交互脚本、2个Python辅助脚本、1个Scala主程序以及README说明文档其中HTML负责展示分析结果JSON提供数据支撑JS完成前端交互Python与Scala处理后台逻辑目录结构清晰便于按模块阅读和二次开发。项目代码经过运行测试功能均可正常使用作者答辩评审平均分达96分如遇运行问题还可私聊联系作者获取远程教学指导。目前已有244人学习下载适合作为业务型数据分析实战练习也适合在现有代码基础上进行功能扩展作为毕业设计或课程设计的可靠参考方案。1. TMDB 电影数据集用 Spark 做分析的几个真实理由TMDB 电影数据集常被当成 Spark 入门的练习题但它并不是一张干净的表。movies_metadata.csv、credits.csv、ratings_small.csv 三个文件凑到一起字段里混着 JSON、空字符串和不规范的日期单靠 pandas 也能处理但一旦分析口径要反复调整DataFrame 的惰性求值和 SQL 表达能力能把这类探索工作拉回结构化流程。这也是“基于 Spark 平台 TMDB 电影数据分析”这个标题真正要回答的问题环境参数怎么备、字段怎么清、聚合怎么做、结果怎么验证。下面五章按一条能本地跑通的分析主线展开从 SparkSession 配置到 Schema 声明再到数据清洗、票房聚合、窗口排名和输出验证每一段都可以直接抄进自己的 Spark 项目里运行。动手的人最好已经搭好一个小规模的 Spark Standalone 或 YARN 客户端开发环境剩下的靠代码说话。2. Spark 环境参数与 Schema 设计读取 TMDB CSV 前的准备数据处理的第一步不是写聚合而是决定读取方式和字段类型。这一章解决两件事SparkSession 的参数怎么给TMDB 原始文件的 Schema 怎么声明。两个问题在几万行数据上容易一带而过等作业交付后才发现结果在类型转换环节已经悄悄错了。2.1 读取 TMDB 三个数据文件前要确定分析边界“电影数据分析”落在哪个粒度会影响后面所有代码。我一般会把第一版范围定为“电影元数据 用户评分”两个文件credits.csv 留作扩展。理由很直接片名、预算、票房、类型、日期、评分都集中在 movies_metadata.csv 一个文件里分析性价比最高credits 一旦要 join 角色、导演关系依赖链会被拉长第一次跑通主线时没必要引入。文件包含内容本次分析中的角色movies_metadata.csv片名、预算、票房、类型、日期、评分主表ratings_small.csv用户对电影的评分记录评分分布辅助credits.csv演员、导演、职位信息后续扩展不参与第一版确定边界之后再决定读取路径。本地开发用file:///home/xxx/tmdb-spark/绝对路径相对路径在 Jupyter 和 spark-submit 之间经常不一致。如果提交到 YARN 模式数据文件要提前放到 HDFS并把读取路径改成hdfs:///data/tmdb/movies_metadata.csv否则 driver 本地能找到文件而 executor 全部报 file not found这是集群模式下最常见的启动失败原因。2.2 手动 Schema 声明代替类型推断的三个理由Spark 的read.csv不指定 schema 时会自动推断对 TMDB 文件这个机制会带来三种典型错误。第一id 列在部分行里有非数字字符推断结果可能变成 DoubleType后续与评分表的整数 movieId join 时永远匹配不上。第二budget 和 revenue 里混着 0、空字符串和上亿的大数推断出来的字段类型要么不够长要么把整列定为 string聚合时还得再转。第三genres 字段是带转义引号的 JSON类型推断只负责最外层的 StringType实际解析仍然要手动声明目标类型。手动声明 schema 后CSV 解析器遇到类型不一致的行时能按既定策略处理而不是自动切换类型。先定义主表结构from pyspark.sql.types import ( StructType, StructField, StringType, LongType, DoubleType, IntegerType ) metadata_schema StructType([ StructField(budget, LongType(), True), StructField(genres, StringType(), True), StructField(homepage, StringType(), True), StructField(id, StringType(), True), StructField(keywords, StringType(), True), StructField(original_language, StringType(), True), StructField(original_title, StringType(), True), StructField(overview, StringType(), True), StructField(popularity, DoubleType(), True), StructField(release_date, StringType(), True), StructField(revenue, LongType(), True), StructField(runtime, IntegerType(), True), StructField(title, StringType(), True), StructField(vote_average, DoubleType(), True), StructField(vote_count, IntegerType(), True) ])声明里特意把 id 保留成 StringType留到清洗阶段统一转数值budget 和 revenue 用 LongType 而不是 IntegerType因为单部电影票房超过 21 亿时 IntegerType 会溢出。release_date 也保留字符串源数据里有年份、斜杠日期和空值三种形态直接定成 DateType 会有一批行变成 null。读入文件metadata_raw ( spark.read .option(header, True) .option(quote, \) .option(escape, \) .option(mode, PERMISSIVE) .schema(metadata_schema) .csv(data/tmdb/movies_metadata.csv) ) print(metadata_raw.count()) metadata_raw.printSchema()这几个 option 作用不同。header 把首行作为列名quote 指定字段包裹符escape 指定字段内的转义符缺少它会直接截断 genres 这类 JSONmode 设为 PERMISSIVE 表示遇到溢出不匹配时用 null 填充而不是让整个作业失败schema 与 mode 配合决定哪些行保留。第一次运行时先 count 和 printSchema确认读入行数没有明显少于源文件行数再往下走。提示如果 count 结果明显小于文件行数优先去查文件里是否有长字段跨行。genres、overview 这类字段被双引号包着时内部若含换行就会出现列错位手动 Schema 也救不回来需要加上.option(multiline, True)重新读取。2.3 SparkSession 的几个运行参数怎么设SparkSession 的配置项不用多针对 TMDB 数据规模调三个就够from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(tmdb-movie-analysis) \ .config(spark.sql.shuffle.partitions, 12) \ .config(spark.sql.autoBroadcastJoinThreshold, 104857600) \ .config(spark.sql.session.timeZone, UTC) \ .getOrCreate()参数说明spark.sql.shuffle.partitions控制 groupBy 和 join 后 shuffle 的分区数。默认值是 200在单机跑 TMDB 这种几万行数据会出现大量空 task设成 12 表示聚合后最多 12 个分区并发计算降低调度开销。spark.sql.autoBroadcastJoinThreshold默认 10MB设成 104857600 字节即 100MB 后连接评分表时有机会自动走 BroadcastHashJoin 而不是 SortMergeJoin。spark.sql.session.timeZone指定会话时区提交节点和集群节点时区不一致时to_date 和 year 的结果可能差一天显式声明可以避免这种偏移。这几个参数里最容易踩坑的是 shuffle.partitions 调太大。在一个只有 4 核的环境设置 200Spark UI 里会出现大量空 task日志看起来是 Failed 但作业仍然成功实际是重试空任务导致的假失败现象。调参之后用同一个聚合查询对比执行时间比凭感觉改配置更可靠。3. DataFrame 清洗 TMDB 原始数据JSON 解析、空值和日期转换读取只是开始TMDB 原始文件里有两类不干净数值列的 0 和 null 混在一起以及 genres 字段里嵌着 JSON 需要按结构提取。这一章处理从原始行到可分析表的整段转换也是标题里“源代码”价值最集中的位置。3.1 字符串 id 转 Long 并清除无法识别行第 2 章把 id 保留成字符串这一节先转数值再过滤from pyspark.sql import functions as F metadata metadata_raw \ .withColumn( movie_id, F.when(F.col(id).rlike(^[0-9]$), F.col(id).cast(long)) .otherwise(F.lit(None)) ) \ .filter(F.col(movie_id).isNotNull()) \ .drop(id) metadata.select(title, movie_id).show(5, truncateFalse) print(有效行数:, metadata.count())逻辑说明rlike(^[0-9]$)是正则匹配只保留纯数字的 id。匹配成功的行被 cast 成 long失败的行补 null随后 filter 丢弃。先正则后转换是为了避免 Spark 对非数字字符串直接 cast 时抛出异常。drop 删除旧的字符串列后续所有 join 统一用 movie_id 作为连接键。执行后如果有效行数比 CSV 原始行数少很多先看被过滤掉的行长什么样用metadata_raw.filter(~F.col(id).rlike(^[0-9]$)).select(id, title).show(10, False)输出样本。多数情况是源数据里的脏行去掉不影响总体分析如果占比超过 5%要回去确认 CSV 文件是不是下载不完整。3.2 预算、票房与利润字段的零值处理budget 和 revenue 为 0 的电影在 TMDB 数据里占相当比例这些不是真实的 0而是公开资料缺失。直接用 0 参与平均票房计算会把均值显著拉低所以分析时按两套口径处理一套保留全量用于统计影片数量另一套过滤营收场景专门计算商业指标。movies_numeric metadata \ .withColumn(budget_num, F.coalesce(F.col(budget), F.lit(0))) \ .withColumn(revenue_num, F.coalesce(F.col(revenue), F.lit(0))) \ .withColumn(profit, F.col(revenue_num) - F.col(budget_num)) movies_business movies_numeric.filter(F.col(revenue_num) 0) \ .filter(F.col(budget_num) 0) print(全量电影:, movies_numeric.count()) print(含预算与票房数据的电影:, movies_business.count())coalesce 从参数里取第一个非 null 值这里把 null 统一补成 0。profit 可能是负数代表亏损。过滤条件拆成两个 filter比直接写 and 表达式更容易维护和替换。打印两个 count 能直观看到还有多少样本可用于商业分析如果只剩几百行就不要用平均值指标改成中位数。注意不要覆盖原来的 budget 和 revenue 列。原始值后面反查数据或调整口径时还要用新增派生列比原地覆盖安全得多。3.3 用 from_json 解析 genres 和 keywords 字段genres 字段形如[{id: 28, name: Action}, {id: 12, name: Adventure}]。要用 from_json 把它解析成结构体数组才能进一步展开from pyspark.sql.types import ArrayType, StructType, StructField, IntegerType genre_schema ArrayType( StructType([ StructField(id, IntegerType(), True), StructField(name, StringType(), True) ]) ) movies_mapped movies_business \ .withColumn( genres_arr, F.from_json(F.col(genres), genre_schema) ) \ .withColumn( genre_names, F.when(F.col(genres_arr).isNotNull(), F.expr(transform(genres_arr, g - g.name))) .otherwise(F.array().cast(arraystring)) ) \ .withColumn(primary_genre, F.element_at(F.col(genre_names), 1)) movies_mapped.select(title, genre_names, primary_genre).show(5, truncateFalse)这三个派生列的作用不同。genres_arr 是结构体数组每个元素有 id 和 namegenre_names 通过 transform 高阶函数取出 name 转换成纯字符串数组为后面的 explode 做准备primary_genre 取数组第一个元素作为主类型。from_json 的两个参数一个是列名一个是目标类型。如果 JSON 字段名或嵌套层级与 schema 不匹配解析结果是整体 null 而不是部分 null。写这段时必须避开一个常见坑目标类型声明成ArrayType(StructType(...))如果误写成单纯的 StructType解析结果对不上数组结构整列全是 null。keywords 字段结构与 genres 完全相同复制同一套声明即可。这里还有个性能细节from_json 代价不低如果清洗时 genres 和 keywords 都要解析别写两遍 withColumn而是用 select 一次性扩展出多列减少对同一条原始 JSON 的重复反序列化。3.4 日期标准化与非法行过滤release_date 中混有yyyy-MM-dd和yyyy/MM/dd两种格式对年份聚合的影响很大movies_final movies_mapped \ .filter(F.col(release_date).isNotNull()) \ .withColumn( release_dt, F.to_date(F.col(release_date), yyyy-MM-dd) ) \ .filter(F.col(release_dt).isNotNull()) \ .withColumn(release_year, F.year(F.col(release_dt)))逻辑说明to_date 按yyyy-MM-dd解析遇到斜杠日期返回 null第二个 filter 把这些行排除保证“有上映年份”的分析建立在真实可解析的日期上。如果业务上必须救回斜杠日期要用 unix_timestamp 分次试探两种格式再取非 null 值不能只改 to_date 的格式参数。过滤时也可以把空日期行保留成一个单独的“unknown”分组但 TMDB 数据里这类行极少直接过滤掉对结论几乎没有影响。4. 票房聚合与窗口排名TMDB 电影数据的多维统计实战清洗后的 movies_final 可以复现三类最常见的分析年度票房走势、类型收益差异、每个类型下的 Top-N 电影。这三个场景分别对应 groupBy 聚合、explode 展开、窗口函数排名是 Spark 数据分析案例中出场频率最高的操作组合。4.1 年度票房聚合与平均分统计口径先按年汇总看数据质量和趋势走向yearly_stats movies_final \ .groupBy(release_year) \ .agg( F.count(movie_id).alias(movie_count), F.sum(revenue_num).alias(total_revenue), F.avg(vote_average).alias(avg_rating), F.round(F.avg(profit), 2).alias(avg_profit) ) \ .orderBy(F.desc(release_year)) yearly_stats.show(10, truncateFalse)这条链里最容易忽视的是 avg_rating 会把 vote_count 为 0 的电影也并进平均。无人评分的电影 vote_average 通常为 0直接把平均值拉得很低。更严谨的口径是先过滤出有足够投票数的电影yearly_rated movies_final \ .filter(F.col(vote_count) 20) \ .groupBy(release_year) \ .agg( F.count(movie_id).alias(rated_movie_count), F.round(F.avg(vote_average), 2).alias(avg_rating), F.percentile_approx(F.col(vote_average), 0.5).alias(median_rating) ) \ .orderBy(F.desc(release_year)) yearly_rated.show(10, truncateFalse)percentile_approx 是近似分位数计算第二个参数 0.5 表示中位数。相比精确百分位在大数据量下快得多在 TMDB 这个规模结果与精确计算基本一致。这里同时保留 avg_rating 和 median_rating是因为电影评分分布是长尾的一部高分小众电影会把平均分拉高而中位数更能反映大众观感。如果某一年 movie_count 和 total_revenue 同时明显下降先检查 release_year 是否大量为 null而不是怀疑聚合语法。4.2 用 explode 展开类型字段做多类型统计类型维度分析需要把一部电影的多类型拆成多行genre_exploded movies_final \ .select( movie_id, title, budget_num, revenue_num, profit, vote_average, genre_names ) \ .withColumn(genre_name, F.explode(F.col(genre_names))) genre_stats genre_exploded \ .groupBy(genre_name) \ .agg( F.count(movie_id).alias(movie_cnt), F.avg(revenue_num).alias(avg_revenue), F.avg(profit).alias(avg_profit), F.avg(vote_average).alias(avg_rating) ) \ .orderBy(F.desc(movie_cnt)) genre_stats.show(20, truncateFalse)explode 把数组的每个元素生成一行因此一部同时属于 Action 和 Adventure 的电影会出现在两个类型分组里。这时的 movie_cnt 语义是“该类型下电影出现次数”不是去重后的电影数量。要表达“有多少部电影属于该类型”先用 movie_id 和 genre_name 做 distinct 再 count。select 里只保留后续需要的列可以减少 explode 之后管道中传输的数据量这个习惯在数据量大时能明显降低 shuffle 成本。看结果时注意 avg_revenue 会被少数商业大片拉高类型间的真实差异建议再看一版中位数。genre_exploded 已经是公共中间结果后面所有类型维度的二次聚合都复用它不再重复 explode。4.3 用窗口函数在类型内部算 Top-N 排名常见的需求是每个类型下票房最高的 10 部电影。窗口函数解决“组内排序后取前 N 条”比 groupBy 之后手工 collect_list 再截断直观很多from pyspark.sql import Window genre_win Window.partitionBy(genre_name).orderBy(F.desc(revenue_num)) top_movies genre_exploded \ .withColumn(rank, F.row_number().over(genre_win)) \ .filter(F.col(rank) 10) \ .select(genre_name, title, revenue_num, vote_average, rank) \ .orderBy(genre_name, rank) top_movies.show(20, truncateFalse)执行逻辑是partitionBy 按 genre_name 分组orderBy 在组内按票房降序row_number 生成从 1 开始的连续编号最终 filter 保留前 10 条。这个写法不会丢其他列也不需要二次 join 回主表。排名函数的选择在票房并列时差别很大。row_number 遇到相同票房会随机给出不同名次rank 和 dense_rank 都会让并列电影拿到相同名次区别在于 rank 会在并列后跳过后续编号。业务上做票房榜单时通常要求并列呈现应优先用 rank再按 title 排序保证输出稳定。5. 缓存、广播连接与结果验证把 Spark 作业收尾的调优手段分析代码跑通之后收尾工作集中在三件容易被跳过的事缓存中间表、用广播替代 shuffle、把关键结果落盘并验证。5.1 缓存选择标准什么时候 cache()什么时候别用对 TMDB 这个规模元数据只有几万行很多时候不需要缓存缓存反而增加 executor 内存压力。判断标准是同一个 DataFrame 是否要被后续两个以上的任务复用。上面生成的 genre_exploded 就是典型它同时服务于类型统计和 Top-N 排名from pyspark import StorageLevel genre_exploded.persist(StorageLevel.MEMORY_AND_DISK) genre_exploded.count()persist 用于指定存储级别MEMORY_AND_DISK 表示内存放不下时落到磁盘。cache() 默认是 MEMORY_ONLY内存不够就重新计算对反复使用的中间结果MEMORY_AND_DISK 更符合本地开发场景。缓存后的第一次 count 才触发真实计算后续查询直接读缓存。源数据逻辑一旦变更记得先 unpersist 再重新 persist不要拿旧缓存当新结果。5.2 用 broadcast 连接评分数据减少 shuffle 成本rating 表只有约十万行非常小适合广播连接。上一章末尾把 autoBroadcastJoinThreshold 调成 100MB这里在代码里显式标记ratings_raw ( spark.read .option(header, True) .csv(data/tmdb/ratings_small.csv) .select( F.col(movieId).cast(long).alias(movie_id), F.col(rating).cast(double).alias(rating) ) .filter(F.col(movie_id).isNotNull()) ) ratings_joined movies_final.join( F.broadcast(ratings_raw), movie_id, inner ) print(ratings_joined.count())F.broadcast 是显式广播标记Spark 会把 ratings_raw 序列化分发到每个 executor对 movies_final 逐条匹配。这样 shuffle 只发生在 movies_final 一侧评分表不参与全量排序。反过来如果小表很大比如超过 100MB 还强制广播会造成 driver 端 GC 压力和 executor 内存溢出不要滥用。可以在 Spark UI 的 SQL 标签页确认 join 节点是不是 BroadcastHashJoin如果不是去检查阈值配置是否生效。5.3 落盘与数值正确性验证最后把统计结果输出成 parquet同时做一次数值校验避免上游清洗逻辑悄悄带歪结果movies_final.write.mode(overwrite).parquet(output/movies_clean) yearly_rated.write.mode(overwrite).parquet(output/yearly_stats) top_movies.write.mode(overwrite).parquet(output/top_movies) check movies_final.agg( F.count(movie_id).alias(total_movies), F.sum(revenue_num).alias(total_revenue), F.avg(vote_average).alias(avg_all_rating) ).collect() assert check[0][0] 3000, 有效电影数过少检查过滤条件 assert check[0][1] 0, 总票房非正检查收入字段类型 print(check)把断言写进脚本而不是只在控制台看一眼是为了让流程可以放进定时任务或 CI。断言阈值要按实际数据定义这里 3000 是 TMDB 全量数据清洗后的合理下限。panel 更精细的验证方式是把聚合结果和原始 CSV 的唯一 id 数对比如果两边数字差太多多半是第 3 章的 id 过滤或日期过滤过于严格。落盘完成后用spark.read.parquet(output/movies_clean).createOrReplaceTempView(movies)注册成临时视图后续任何聚合结果都可以用一行 SQL 快速反查这是调整分析参数时最省事的验证手段。本文还有配套的精品资源点击获取
)
)
