前阵子一直在补 Spark 的实操想找个有一定规模、字段又足够丰富的公开数据集练手翻了很多方向最后锁定了 TMDB 电影数据。这个数据源在影视分析里很常见里面有电影评分、票房、制作预算、类型、演员、关键词等一堆信息用 Spark Scala 跑一遍能顺手把 DataFrame 的读取、清洗、聚合、窗口函数、JSON 解析这些基本功全部过一遍。这篇文章就按实际项目顺序整理出来包含完整可跑的 Scala 代码和开发期碰到的真问题。适合刚接触 Spark SQL、想拿真实数据做练习的同学也适合要交大数据课程作业的朋友直接参考。1. 项目概述与整体设计先搞清楚数据再写代码很多人在拿到数据集之后第一反应就是开 IDE 写代码我见过不少一上来就把 CSV 读进来然后各种 show最后发现字段解读错了整个分析推倒重来。这个项目的第一步不是写 Spark而是先搞清楚 TMDB 数据里有什么、能算什么指标、最后要输出什么。1.1 TMDB数据里到底有什么字段TMDB 这套电影数据通常拆成两个 CSV 文件一个是电影基本信息另一个是演员和制作团队信息。电影信息表是每部电影一行包含预算、收入、口碑、热度、时长、语言、类型、关键词等字段。另一个 credits 表也是每部电影一行但里面的 cast 和 crew 不是普通字符串而是 JSON 数组格式的字符串每个成员包含演员名字、角色名、出场顺序这类信息。文件主要字段字段说明movies.csvbudget / revenue预算和票房收入单位是美元movies.csvvote_average / vote_count用户平均评分和投票总数movies.csvpopularityTMDB 计算的热度值movies.csvgenres / keywordsJSON 数组字符串需要解析后才能分析movies.csvrelease_date / runtime上映日期和片长movies.csvoriginal_language原始语言代码credits.csvmovie_id关联 movies.csv 的电影 IDcredits.csvcastJSON 数组成员是演员信息credits.csvcrewJSON 数组成员是导演、编剧等这份数据大约有四千多部电影量级不算大但字段结构很有层次感尤其是 genres、keywords、cast 这种半结构化嵌套字段跟真实生产环境里的日志、埋点、订单快照非常像。Spark 最擅长处理的就是这种场景数据量大、结构复杂、需要分布式清洗和聚合。1.2 为什么用SparkScala而不是Pandas有人可能会问四千多部电影用 Pandas 读进内存也就几秒为什么要折腾 Spark。这个问题的关键不在数据量而在处理方式和方法论。Pandas 适合单机交互式分析但数据一旦涨到几个 GB 或者几十 GB内存很容易爆掉而且代码迁移到生产集群时要重写。Spark 的核心优势是分布式计算、列式存储格式、SQL 接口和成熟的算子体系写完的 DataFrame 逻辑可以直接从本地切到集群跑。至于为什么选 Scala而不是更流行的 Python API。Spark 本身就是 Scala 写的使用 Scala 能直接调用最原生的 API类型更严谨IDE 里的自动补全和编译期检查也能帮你提前发现很多错误。对工程化项目来说Scala 版的 Spark 代码在生产环境里非常常见。缺点当然也有Scala 写起来比 Python 绕可视化也不如 Python 方便所以这个项目里我只用 Spark 算指标结果导出来之后再交给其他工具展示。1.3 分析流程从四个维度拆解这个项目我拆成了四步数据接入与摸底数据清洗与宽表构建核心指标分析进阶标签分析。核心指标围绕口碑、票房、热度三个维度展开进阶分析围绕类型、演员、关键词展开。为什么先做宽表再做分析因为 from_json 解析是有开销的如果在每个分析里都重新解析一次 genres、cast、keywords既浪费计算资源也让代码变得啰嗦。正确做法是先把 CSV 清洗成一张 Parquet 宽表把嵌套字段解析成数组列之后所有分析都读这张宽表速度会快很多。2. 环境准备与工程骨架版本组合锁死做 Spark 项目第一件事不是写业务代码而是确定版本组合。Spark、Scala、JDK 三个版本只要有一个不匹配编译能过运行也会给你来一段莫名其妙的序列化错误。与其事后排查不如一开始就锁死版本。2.1 Spark 3.3 Scala 2.12的依赖怎么配我用的组合是 Spark 3.3.2、Scala 2.12.15、JDK 8/11、sbt 1.7。Spark 3.x 准备了两套 Scala 编译版本2.12 和 2.13实际生产里 2.12 的兼容性和第三方库成熟度更高所以我选 2.12。JDK 版本不建议追新JDK 17 在某些 Spark 版本上有模块访问问题老老实实用 8 或 11 最稳。sbt项目核心配置如下name : tmdb-analysis version : 1.0 scalaVersion : 2.12.15 libraryDependencies Seq( org.apache.spark %% spark-core % 3.3.2, org.apache.spark %% spark-sql % 3.3.2 )这里有一个容易忽略的点本地用 sbt run 跑没问题但如果要发到集群上用 spark-submit 提交建议在 dependencies 中使用provided避免把 Spark 自带的 jar 打进去导致冲突。另一种方案是打成 assembly 包但需要把 spark-core、spark-sql 排除掉这个实操里很常见。2.2 本地调试用的SparkSession参数SparkSession 是入口配置参数直接决定本地调试体验。默认的 Spark 参数是为生产集群设计的比如 spark.sql.shuffle.partitions 默认是 200本地几千条数据跑 groupBy硬生生会产生 200 个小分区纯属浪费。本地调试我一般这样配置val spark SparkSession.builder() .appName(TMDB Movie Analysis) .master(local[2]) .config(spark.sql.shuffle.partitions, 8) .getOrCreate() spark.sparkContext.setLogLevel(WARN)解释一下local[2]这是让 Spark 使用两个本地线程也就是模拟两个 executor 核心。不要一上来就local[*]把机器所有核全部占满尤其是你还开着浏览器、IDE 的时候机器会卡到怀疑人生。shuffle.partitions调成 8是因为本地数据量小8 个 reduce 任务足够用了。生产环境里的设置逻辑又不一样那是另一个话题后面调优部分再展开。2.3 先摸一遍Schema和样例数据环境准备好之后先别急着写分析逻辑先做数据摸底。我用 Spark 的 CSV 读取器把电影表和演员表加载进来然后打印 Schema 和前几行数据。val movies spark.read .option(header, true) .option(multiLine, true) .option(escape, \) .option(inferSchema, true) .csv(data/tmdb_5000_movies.csv) movies.printSchema() movies.select(title, release_date, budget, revenue, vote_average, genres).show(3)打印出来的 Schema 基本符合预期但有一个非常关键的发现genres 这一列虽然是 JSON 数组字符串但 Spark 的 inferSchema 只把它识别成 StringType不会自动解析成数组。这决定了后续必须手动用 from_json 处理。release_date 也大概率是 StringType后面用 to_date 统一转换。先把这些问题摸清楚后面的清洗逻辑心里就有底了。3. 数据清洗与预处理把半结构化数据变成宽表数据清洗是整个项目里耗时最长、最烦的部分。TMDB 的原始 CSV 有个特点很多字段是 JSON 字符串而且 CSV 里还有多行文本和转义引号。如果不处理干净后面每个分析都会被脏数据污染。3.1 多行CSV读取容易踩的坑TMDB 的 movies.csv 里有一个 overview 字段电影简介内容很长里面可能有换行符和逗号。Spark 默认的 CSV 解析器按行读取碰到换行就直接切断记录结果就是字段错位、行数变多、数据乱七八糟。解决办法是加两个 optionmultiLine 和 escape。val credits spark.read .option(header, true) .option(multiLine, true) .option(escape, \) .option(inferSchema, true) .csv(data/tmdb_5000_credits.csv)multiLine 让解析器支持字段内部换行escape 告诉解析器用双引号作为转义字符。第一次跑的时候我只加了 header结果正常记录不到一半排查了大半天才意识到是多行 CSV 的问题。这个坑非常典型如果你们拿到的数据集包含长文本读取时一定要先确认这三件套multiLine、escape、header。3.2 JSON字符串拆成结构化字段from_json配合explodeTMDB 电影表里的 genres 字段长这样[{id:28,name:Action},{id:12,name:Adventure}]这是一个数组每个元素是一个对象。如果用纯字符串处理想统计类型分布就只能做正则匹配又慢又容易出错。正确做法是用 Spark SQL 的 from_json 函数把它解析成一个数组列。import org.apache.spark.sql.types._ import org.apache.spark.sql.functions._ val genreSchema ArrayType( StructType(Seq( StructField(id, IntegerType), StructField(name, StringType) )) ) val moviesWithGenres movies .withColumn(genres_arr, from_json($genres, genreSchema)) .withColumn(genre_names, $genres_arr.name)解析之后genre_names 是一个 ArrayType(StringType) 的列里面装着该电影的所有类型名。这样后面想统计单个类型就 explode 它想保留整组类型就直接用数组列参与计算非常灵活。这里有个重要经验解析 JSON 字段时schema 定义要和数据完全对得上。比如 genres 的字段是 id 和 name你定义成 id 和 title解析结果就会全是 null。Spark 的 from_json 解析失败不会报错而是静默返回 null这一步特别坑。所以解析完以后要过滤一下确认非空行比例正常。3.3 清洗后的宽表设计与Parquet落盘清洗阶段的最终目标不是单纯去 null而是构造一张后续所有分析都能直接用的宽表。我把日期转换、年份提取、预算和收入转成长整型、JSON 数组解析都放在一个流程里完成。val moviesClean movies .filter($id.isNotNull) .withColumn(release_date_ts, to_date($release_date, yyyy-MM-dd)) .filter($release_date_ts.isNotNull) .withColumn(release_year, year($release_date_ts)) .withColumn(budget_long, $budget.cast(long)) .withColumn(revenue_long, $revenue.cast(long)) .withColumn(runtime_int, $runtime.cast(int)) .withColumn(genres_arr, from_json($genres, genreSchema)) .withColumn(genre_names, $genres_arr.name) .withColumn(keywords_arr, from_json($keywords, keywordSchema)) .withColumn(keyword_names, $keywords_arr.name) .select( $id.as(movie_id), $title, $release_year, $runtime_int.as(runtime), $budget_long.as(budget), $revenue_long.as(revenue), $popularity, $vote_average, $vote_count, $genre_names, $keyword_names ) moviesClean.write .mode(overwrite) .partitionBy(release_year) .parquet(output/tmdb_clean)落盘用 Parquet 而不是普通 CSV是因为 Parquet 是列式存储后续分析按年份筛选时只需要读取相关列I/O 开销小很多。分区字段就用 release_year按年份组织文件这也是模拟真实数仓里的分区表设计思路。4. 核心分析口碑、票房与热度三维度宽表建好之后真正的分析才刚开始。这个项目里我从口碑、票房、热度三个维度切入每个维度又分别补充了一两个更细化的指标这样整个分析才有层次。4.1 年度评分趋势聚合窗口滑动平均口碑分析最直观的指标是年度平均评分。为了防止只有几个人投票的电影干扰结果我这里过滤掉 vote_count 小于 50 的样本只保留有一定投票基础的电影。val yearlyStats spark.read.parquet(output/tmdb_clean) .filter($vote_count 50) .groupBy(release_year) .agg( count(*).alias(movie_count), avg(vote_average).alias(avg_rating), stddev(vote_average).alias(rating_stddev) ) .orderBy($release_year.desc) yearlyStats.show(20)聚合结果能看出每年上映电影数量和平均评分的变化轨迹。如果只看每年的离散值趋势还不够平滑所以我又加了一个窗口函数计算三年滑动平均。import org.apache.spark.sql.expressions.Window val windowSpec Window.orderBy($release_year).rowsBetween(-1, 1) val yearlyWithTrend yearlyStats .withColumn(avg_rating_ma, avg(avg_rating).over(windowSpec))窗口函数和 groupBy 的区别在于groupBy 会把多行压成一行窗口函数不会减少行数而是在每一行周围扩展一个窗口做计算。这里用 rowsBetween(-1, 1) 表示当前年份、上一年、下一年的平均值。做时间序列分析时这种滑动平均比原始值更容易看出整体趋势。4.2 票房榜与投资回报率怎么算更合理票房分析不能只看 revenue还要结合 budget 看利润和回报率。利润是 revenue 减 budget回报率就是利润除以预算。直接算会遇到一些极端值有些电影预算极低、收入几十万美元ROI 看上去高到离谱但实际上没什么参考价值。所以我过滤掉收入或预算为 0 的数据同时也过滤掉投票数太少的冷门影片。val finance spark.read.parquet(output/tmdb_clean) .filter($budget 0) .filter($revenue 0) .filter($vote_count 20) .withColumn(profit, $revenue - $budget) .withColumn(roi, ($revenue - $budget) / $budget) .select(title, release_year, budget, revenue, profit, roi) finance.orderBy($profit.desc).show(15) finance.orderBy($roi.desc).show(15)这里能给一个很实用的建议分析商业表现时利润榜和 ROI 榜要分开看。利润榜容易出大制作、高投入的头部影片ROI 榜更容易出现以小博大的黑马。两者结合才能避免“只看赚得多不看赚得快”的偏差。在实际跑数据时还有个容易犯的错直接用原始 revenue 和 budget 字段这两个字段在 CSV 里可能是字符串也可能是 null。虽然 inferSchema 可能推断成整型但保险起见在宽表阶段已经 cast 成 long这一步就省心很多。4.3 热度分层用分位数划分后再看评分popularity 字段是一个热度值这个值的分布通常是偏态的少数头部电影热度极高大多数尾部电影热度很低。用固定阈值比如“50”去分高低并不合理更好的办法是看分位数。val base spark.read.parquet(output/tmdb_clean).cache() val quantiles base.stat.approxQuantile(popularity, Array(0.4, 0.7, 0.9), 0.01) val q40 quantiles(0) val q70 quantiles(1) val q90 quantiles(2) val heatLevel base .withColumn(heat_level, when($popularity q90, High) .when($popularity q70, Medium-High) .when($popularity q40, Medium) .otherwise(Low) ) .groupBy(heat_level) .agg( count(*).alias(movie_count), avg(vote_average).alias(avg_rating) ) .orderBy($avg_rating.desc) heatLevel.show()approxQuantile 是 Spark 提供的近似分位数计算接口第三个参数是相对精度 0.01表示允许 1% 的误差。这样分层的好处是能观察不同热度区间下的平均评分差异。从实际数据看高热度不一定等于高评分有些低热度冷门片评分反而很高所以热度和口碑是两个独立维度不能简单混着看。这个结论也提醒我们在后续分析里一定要区分“叫好”和“卖座”。5. 进阶分析类型、演员和关键词基础指标看完了再把内容标签类数据利用起来。类型、演员、关键词这三类信息都能拆成一个个标签属于典型的“宽表转长表”分析也是 Spark 里 explode 最适用的场景。5.1 类型分布与类型组合TopN统计类型分布非常简单对 genre_names 数组列做 explode然后按类型名分组计数。val genreCounts base .filter($genre_names.isNotNull) .withColumn(genre, explode($genre_names)) .groupBy(genre) .agg(count(*).alias(movie_count)) .orderBy($movie_count.desc) genreCounts.show(15)但这里有一个细节一部电影可以属于多个类型所以 type 分布里同一部电影可能在 Drama 和 Thriller 里各计一次这不算 bug但解读结果时要清楚“计数”是类型出现次数不是电影数量。比单类型分布更有意思的是类型组合。很多电影不是只有一个类型而是几个类型同时存在比如“动作冒险”“剧情爱情”。如果直接拼接类型名顺序不同会导致组合被拆成两行。我的办法是先对数组做 sort_array 排序再拼接成 key这样“ActionDrama”和“DramaAction”会归到同一个组合。val genreCombos base .filter($genre_names.isNotNull) .groupBy(title) .agg(sort_array($genre_names).alias(genres_sorted)) .withColumn(combo, concat_ws(, $genres_sorted)) .groupBy(combo) .agg(count(*).alias(movie_count)) .orderBy($movie_count.desc) genreCombos.show(10)排序后再拼接这个思路在处理所有无序标签组合时都非常实用不只是电影类型。5.2 演员作品量和平均分explode后join演员信息在 credits 表的 cast 字段里结构比 genres 更复杂。这里需要定义 cast 的 JSON schema然后解析成数组再 explode 成演员维度。val castSchema ArrayType(StructType(Seq( StructField(cast_id, IntegerType), StructField(character, StringType), StructField(credit_id, StringType), StructField(gender, IntegerType), StructField(id, IntegerType), StructField(name, StringType), StructField(order, IntegerType) ))) val creditsParsed credits .withColumn(cast_arr, from_json($cast, castSchema)) .withColumn(actor, explode_outer($cast_arr)) .withColumn(actor_name, $actor.name) .select($movie_id.as(id), $actor_name)注意这里我用的是 explode_outer而不是 explode。区别在于如果某部电影的 cast 字段解析后是 nullexplode 会让这一行消失explode_outer 会把这一行保留下来actor_name 为 null。对演员分析来说丢几部其实影响不大但在生产环境里保留 null 往往更有价值因为你能统计出有多少数据是缺失的。接下来把演员表和宽表按电影 ID join然后按演员聚合val actorStats creditsParsed .join(base.select(id, vote_average, revenue), Seq(id), inner) .groupBy(actor_name) .agg( count(*).alias(movie_count), avg(vote_average).alias(avg_rating), sum(revenue).alias(total_revenue) ) .filter($movie_count 5) .orderBy($movie_count.desc, $avg_rating.desc) actorStats.show(15)join 之后还按 movie_count 过滤是为了排除只演过一两部电影的偶然性。要分析“稳定产出”的演员样本量太少没有意义。这个模块跑完既能找出作品数量最多的演员也能找出高口碑演员还能看出累计票房最高的演员是谁。5.3 关键词高频统计关键词字段在 movies.csv 的 keywords 列同样是一组 JSON 数组。解析方式和 genres 一模一样这也是为什么我在宽表阶段就把 keyword_names 建出来的原因。val keywordStats spark.read.parquet(output/tmdb_clean) .filter($keyword_names.isNotNull) .withColumn(kw, explode($keyword_names)) .groupBy(kw) .agg(count(*).alias(movie_count)) .orderBy($movie_count.desc) keywordStats.show(30)关键词统计可以反映某种内容主题的流行程度。比如“独立电影”“女性导演”“反乌托邦”这类标签出现频率越高说明这个主题在样本里的覆盖度越大。和类型组合一样关键词也属于无序标签如果有需要做组合分析同样要先 sort_array 再拼接。这套方法在标签系统、内容分类、用户画像场景里几乎可以原样复用。6. 开发期常见问题与调优实录最后一部分聊开发过程中真实遇到的坑。Spark 项目最耗时间的不是写逻辑而是处理数据格式、资源分配和性能问题。我把最有代表性的几个问题整理成排查记录给后来人做参考。6.1 CSV错位、日期空值和JSON解析失败第一个大坑就是 CSV 里带换行的长文本。如果没开 multiLineSpark 会把一条电影记录拆成多条导致 title、budget、genres 全部错位。排查方法很简单读进来之后先 count如果行数明显大于电影数量九成就是多行 CSV 没处理对。第二个坑是日期解析。release_date 列里有正常日期也有空字符串和脏格式。to_date 解析不认识的格式会直接返回 null不会报错。不少人看到 null 以为是函数有问题其实是被脏数据拖累了。我的处理方式是解析之后 filter 掉 null然后再提取年份。第三个坑是 from_json 解析失败静默返回 null。如果 genres_arr 大量为 null大概率是 schema 定义和数据不匹配。这时不要盲目改代码先拿一条原始 JSON 出来对照看字段名到底是 id/name还是其他名字。这个习惯能帮你省下好几个小时。6.2 数据倾斜怎么定位加盐两阶段聚合示例做演员聚合时容易遇到数据倾斜问题。比如一个热门演员出演了几百部电影按演员分组后包含这个演员的 task 要比其他 task 多处理很多数据导致整个 Stage 卡在那一个 task 上。定位方法很简单打开 Spark UI 看 Stage 页如果某个 task 的 Shuffle Read 或处理时间明显高于 others就需要处理倾斜。常用手段是加盐两阶段聚合。第一轮给 key 加一个随机后缀拆分大 key第二轮把后缀去掉再做一次聚合。val salted creditsParsed .withColumn(salt, (rand() * 20).cast(int)) .groupBy(actor_name, salt) .agg(count(*).alias(partial_cnt)) val result salted .groupBy(actor_name) .agg(sum(partial_cnt).alias(movie_count))需要注意如果还要计算平均分这类指标不能简单用这种 count 聚合。要做加权平均第一轮同时算出 count 和 avg第二轮用 sum(count * avg) / sum(count)。加盐不是越多越好盐值太大会引入额外的 shuffle 开销。小数据集和本机环境一般不会遇到明显倾斜但把方法掌握好生产环境里碰到大数据量就能直接套用。6.3 资源参数、落盘和Spark开发习惯开发期一定要调整 Spark 默认参数。除了上文说的本地 shuffle.partitions还要注意别在小数据集上滥用大配置。你有多少数据量就配多少资源否则只是增加调度开销。结果输出时不要一上来就 collect。collect 会把所有数据拉到 driver 端数据量一大直接 OOM。正确做法是先用 show 看几条确认无误后写到文件。如果输出文件很小可以用 coalesce(1) 合并成一个文件避免生成大量碎片文件。如果后续还要反复读取优先保存 Parquet不要保存 CSV。重复读 Parquet 比重新解析 CSV 快得多。另外Spark UI 是排查性能问题的第一现场。任务执行时间长先看是不是有数据倾斜提示内存不足先看是 driver 还是 executor再决定是调大内存还是优化 shuffle。不要一碰到 OOM 就盲目调参先看数据特征很多问题其实是代码写法造成的。把这套流程跑通之后再面对更大规模的数据无非是把 local 换成集群、把 CSV 换成分区表分析套路是通用的。写到最后分享一个我自己的习惯任何聚合结果先用 show 确认行数和采样再决定要不要落到磁盘千万别一上来就 collect 到本地这大概是 Spark 开发里性价比最高的一条经验。
前阵子一直在补 Spark 的实操想找个有一定规模、字段又足够丰富的公开数据集练手翻了很多方向最后锁定了 TMDB 电影数据。这个数据源在影视分析里很常见里面有电影评分、票房、制作预算、类型、演员、关键词等一堆信息用 Spark Scala 跑一遍能顺手把 DataFrame 的读取、清洗、聚合、窗口函数、JSON 解析这些基本功全部过一遍。这篇文章就按实际项目顺序整理出来包含完整可跑的 Scala 代码和开发期碰到的真问题。适合刚接触 Spark SQL、想拿真实数据做练习的同学也适合要交大数据课程作业的朋友直接参考。1. 项目概述与整体设计先搞清楚数据再写代码很多人在拿到数据集之后第一反应就是开 IDE 写代码我见过不少一上来就把 CSV 读进来然后各种 show最后发现字段解读错了整个分析推倒重来。这个项目的第一步不是写 Spark而是先搞清楚 TMDB 数据里有什么、能算什么指标、最后要输出什么。1.1 TMDB数据里到底有什么字段TMDB 这套电影数据通常拆成两个 CSV 文件一个是电影基本信息另一个是演员和制作团队信息。电影信息表是每部电影一行包含预算、收入、口碑、热度、时长、语言、类型、关键词等字段。另一个 credits 表也是每部电影一行但里面的 cast 和 crew 不是普通字符串而是 JSON 数组格式的字符串每个成员包含演员名字、角色名、出场顺序这类信息。文件主要字段字段说明movies.csvbudget / revenue预算和票房收入单位是美元movies.csvvote_average / vote_count用户平均评分和投票总数movies.csvpopularityTMDB 计算的热度值movies.csvgenres / keywordsJSON 数组字符串需要解析后才能分析movies.csvrelease_date / runtime上映日期和片长movies.csvoriginal_language原始语言代码credits.csvmovie_id关联 movies.csv 的电影 IDcredits.csvcastJSON 数组成员是演员信息credits.csvcrewJSON 数组成员是导演、编剧等这份数据大约有四千多部电影量级不算大但字段结构很有层次感尤其是 genres、keywords、cast 这种半结构化嵌套字段跟真实生产环境里的日志、埋点、订单快照非常像。Spark 最擅长处理的就是这种场景数据量大、结构复杂、需要分布式清洗和聚合。1.2 为什么用SparkScala而不是Pandas有人可能会问四千多部电影用 Pandas 读进内存也就几秒为什么要折腾 Spark。这个问题的关键不在数据量而在处理方式和方法论。Pandas 适合单机交互式分析但数据一旦涨到几个 GB 或者几十 GB内存很容易爆掉而且代码迁移到生产集群时要重写。Spark 的核心优势是分布式计算、列式存储格式、SQL 接口和成熟的算子体系写完的 DataFrame 逻辑可以直接从本地切到集群跑。至于为什么选 Scala而不是更流行的 Python API。Spark 本身就是 Scala 写的使用 Scala 能直接调用最原生的 API类型更严谨IDE 里的自动补全和编译期检查也能帮你提前发现很多错误。对工程化项目来说Scala 版的 Spark 代码在生产环境里非常常见。缺点当然也有Scala 写起来比 Python 绕可视化也不如 Python 方便所以这个项目里我只用 Spark 算指标结果导出来之后再交给其他工具展示。1.3 分析流程从四个维度拆解这个项目我拆成了四步数据接入与摸底数据清洗与宽表构建核心指标分析进阶标签分析。核心指标围绕口碑、票房、热度三个维度展开进阶分析围绕类型、演员、关键词展开。为什么先做宽表再做分析因为 from_json 解析是有开销的如果在每个分析里都重新解析一次 genres、cast、keywords既浪费计算资源也让代码变得啰嗦。正确做法是先把 CSV 清洗成一张 Parquet 宽表把嵌套字段解析成数组列之后所有分析都读这张宽表速度会快很多。2. 环境准备与工程骨架版本组合锁死做 Spark 项目第一件事不是写业务代码而是确定版本组合。Spark、Scala、JDK 三个版本只要有一个不匹配编译能过运行也会给你来一段莫名其妙的序列化错误。与其事后排查不如一开始就锁死版本。2.1 Spark 3.3 Scala 2.12的依赖怎么配我用的组合是 Spark 3.3.2、Scala 2.12.15、JDK 8/11、sbt 1.7。Spark 3.x 准备了两套 Scala 编译版本2.12 和 2.13实际生产里 2.12 的兼容性和第三方库成熟度更高所以我选 2.12。JDK 版本不建议追新JDK 17 在某些 Spark 版本上有模块访问问题老老实实用 8 或 11 最稳。sbt项目核心配置如下name : tmdb-analysis version : 1.0 scalaVersion : 2.12.15 libraryDependencies Seq( org.apache.spark %% spark-core % 3.3.2, org.apache.spark %% spark-sql % 3.3.2 )这里有一个容易忽略的点本地用 sbt run 跑没问题但如果要发到集群上用 spark-submit 提交建议在 dependencies 中使用provided避免把 Spark 自带的 jar 打进去导致冲突。另一种方案是打成 assembly 包但需要把 spark-core、spark-sql 排除掉这个实操里很常见。2.2 本地调试用的SparkSession参数SparkSession 是入口配置参数直接决定本地调试体验。默认的 Spark 参数是为生产集群设计的比如 spark.sql.shuffle.partitions 默认是 200本地几千条数据跑 groupBy硬生生会产生 200 个小分区纯属浪费。本地调试我一般这样配置val spark SparkSession.builder() .appName(TMDB Movie Analysis) .master(local[2]) .config(spark.sql.shuffle.partitions, 8) .getOrCreate() spark.sparkContext.setLogLevel(WARN)解释一下local[2]这是让 Spark 使用两个本地线程也就是模拟两个 executor 核心。不要一上来就local[*]把机器所有核全部占满尤其是你还开着浏览器、IDE 的时候机器会卡到怀疑人生。shuffle.partitions调成 8是因为本地数据量小8 个 reduce 任务足够用了。生产环境里的设置逻辑又不一样那是另一个话题后面调优部分再展开。2.3 先摸一遍Schema和样例数据环境准备好之后先别急着写分析逻辑先做数据摸底。我用 Spark 的 CSV 读取器把电影表和演员表加载进来然后打印 Schema 和前几行数据。val movies spark.read .option(header, true) .option(multiLine, true) .option(escape, \) .option(inferSchema, true) .csv(data/tmdb_5000_movies.csv) movies.printSchema() movies.select(title, release_date, budget, revenue, vote_average, genres).show(3)打印出来的 Schema 基本符合预期但有一个非常关键的发现genres 这一列虽然是 JSON 数组字符串但 Spark 的 inferSchema 只把它识别成 StringType不会自动解析成数组。这决定了后续必须手动用 from_json 处理。release_date 也大概率是 StringType后面用 to_date 统一转换。先把这些问题摸清楚后面的清洗逻辑心里就有底了。3. 数据清洗与预处理把半结构化数据变成宽表数据清洗是整个项目里耗时最长、最烦的部分。TMDB 的原始 CSV 有个特点很多字段是 JSON 字符串而且 CSV 里还有多行文本和转义引号。如果不处理干净后面每个分析都会被脏数据污染。3.1 多行CSV读取容易踩的坑TMDB 的 movies.csv 里有一个 overview 字段电影简介内容很长里面可能有换行符和逗号。Spark 默认的 CSV 解析器按行读取碰到换行就直接切断记录结果就是字段错位、行数变多、数据乱七八糟。解决办法是加两个 optionmultiLine 和 escape。val credits spark.read .option(header, true) .option(multiLine, true) .option(escape, \) .option(inferSchema, true) .csv(data/tmdb_5000_credits.csv)multiLine 让解析器支持字段内部换行escape 告诉解析器用双引号作为转义字符。第一次跑的时候我只加了 header结果正常记录不到一半排查了大半天才意识到是多行 CSV 的问题。这个坑非常典型如果你们拿到的数据集包含长文本读取时一定要先确认这三件套multiLine、escape、header。3.2 JSON字符串拆成结构化字段from_json配合explodeTMDB 电影表里的 genres 字段长这样[{id:28,name:Action},{id:12,name:Adventure}]这是一个数组每个元素是一个对象。如果用纯字符串处理想统计类型分布就只能做正则匹配又慢又容易出错。正确做法是用 Spark SQL 的 from_json 函数把它解析成一个数组列。import org.apache.spark.sql.types._ import org.apache.spark.sql.functions._ val genreSchema ArrayType( StructType(Seq( StructField(id, IntegerType), StructField(name, StringType) )) ) val moviesWithGenres movies .withColumn(genres_arr, from_json($genres, genreSchema)) .withColumn(genre_names, $genres_arr.name)解析之后genre_names 是一个 ArrayType(StringType) 的列里面装着该电影的所有类型名。这样后面想统计单个类型就 explode 它想保留整组类型就直接用数组列参与计算非常灵活。这里有个重要经验解析 JSON 字段时schema 定义要和数据完全对得上。比如 genres 的字段是 id 和 name你定义成 id 和 title解析结果就会全是 null。Spark 的 from_json 解析失败不会报错而是静默返回 null这一步特别坑。所以解析完以后要过滤一下确认非空行比例正常。3.3 清洗后的宽表设计与Parquet落盘清洗阶段的最终目标不是单纯去 null而是构造一张后续所有分析都能直接用的宽表。我把日期转换、年份提取、预算和收入转成长整型、JSON 数组解析都放在一个流程里完成。val moviesClean movies .filter($id.isNotNull) .withColumn(release_date_ts, to_date($release_date, yyyy-MM-dd)) .filter($release_date_ts.isNotNull) .withColumn(release_year, year($release_date_ts)) .withColumn(budget_long, $budget.cast(long)) .withColumn(revenue_long, $revenue.cast(long)) .withColumn(runtime_int, $runtime.cast(int)) .withColumn(genres_arr, from_json($genres, genreSchema)) .withColumn(genre_names, $genres_arr.name) .withColumn(keywords_arr, from_json($keywords, keywordSchema)) .withColumn(keyword_names, $keywords_arr.name) .select( $id.as(movie_id), $title, $release_year, $runtime_int.as(runtime), $budget_long.as(budget), $revenue_long.as(revenue), $popularity, $vote_average, $vote_count, $genre_names, $keyword_names ) moviesClean.write .mode(overwrite) .partitionBy(release_year) .parquet(output/tmdb_clean)落盘用 Parquet 而不是普通 CSV是因为 Parquet 是列式存储后续分析按年份筛选时只需要读取相关列I/O 开销小很多。分区字段就用 release_year按年份组织文件这也是模拟真实数仓里的分区表设计思路。4. 核心分析口碑、票房与热度三维度宽表建好之后真正的分析才刚开始。这个项目里我从口碑、票房、热度三个维度切入每个维度又分别补充了一两个更细化的指标这样整个分析才有层次。4.1 年度评分趋势聚合窗口滑动平均口碑分析最直观的指标是年度平均评分。为了防止只有几个人投票的电影干扰结果我这里过滤掉 vote_count 小于 50 的样本只保留有一定投票基础的电影。val yearlyStats spark.read.parquet(output/tmdb_clean) .filter($vote_count 50) .groupBy(release_year) .agg( count(*).alias(movie_count), avg(vote_average).alias(avg_rating), stddev(vote_average).alias(rating_stddev) ) .orderBy($release_year.desc) yearlyStats.show(20)聚合结果能看出每年上映电影数量和平均评分的变化轨迹。如果只看每年的离散值趋势还不够平滑所以我又加了一个窗口函数计算三年滑动平均。import org.apache.spark.sql.expressions.Window val windowSpec Window.orderBy($release_year).rowsBetween(-1, 1) val yearlyWithTrend yearlyStats .withColumn(avg_rating_ma, avg(avg_rating).over(windowSpec))窗口函数和 groupBy 的区别在于groupBy 会把多行压成一行窗口函数不会减少行数而是在每一行周围扩展一个窗口做计算。这里用 rowsBetween(-1, 1) 表示当前年份、上一年、下一年的平均值。做时间序列分析时这种滑动平均比原始值更容易看出整体趋势。4.2 票房榜与投资回报率怎么算更合理票房分析不能只看 revenue还要结合 budget 看利润和回报率。利润是 revenue 减 budget回报率就是利润除以预算。直接算会遇到一些极端值有些电影预算极低、收入几十万美元ROI 看上去高到离谱但实际上没什么参考价值。所以我过滤掉收入或预算为 0 的数据同时也过滤掉投票数太少的冷门影片。val finance spark.read.parquet(output/tmdb_clean) .filter($budget 0) .filter($revenue 0) .filter($vote_count 20) .withColumn(profit, $revenue - $budget) .withColumn(roi, ($revenue - $budget) / $budget) .select(title, release_year, budget, revenue, profit, roi) finance.orderBy($profit.desc).show(15) finance.orderBy($roi.desc).show(15)这里能给一个很实用的建议分析商业表现时利润榜和 ROI 榜要分开看。利润榜容易出大制作、高投入的头部影片ROI 榜更容易出现以小博大的黑马。两者结合才能避免“只看赚得多不看赚得快”的偏差。在实际跑数据时还有个容易犯的错直接用原始 revenue 和 budget 字段这两个字段在 CSV 里可能是字符串也可能是 null。虽然 inferSchema 可能推断成整型但保险起见在宽表阶段已经 cast 成 long这一步就省心很多。4.3 热度分层用分位数划分后再看评分popularity 字段是一个热度值这个值的分布通常是偏态的少数头部电影热度极高大多数尾部电影热度很低。用固定阈值比如“50”去分高低并不合理更好的办法是看分位数。val base spark.read.parquet(output/tmdb_clean).cache() val quantiles base.stat.approxQuantile(popularity, Array(0.4, 0.7, 0.9), 0.01) val q40 quantiles(0) val q70 quantiles(1) val q90 quantiles(2) val heatLevel base .withColumn(heat_level, when($popularity q90, High) .when($popularity q70, Medium-High) .when($popularity q40, Medium) .otherwise(Low) ) .groupBy(heat_level) .agg( count(*).alias(movie_count), avg(vote_average).alias(avg_rating) ) .orderBy($avg_rating.desc) heatLevel.show()approxQuantile 是 Spark 提供的近似分位数计算接口第三个参数是相对精度 0.01表示允许 1% 的误差。这样分层的好处是能观察不同热度区间下的平均评分差异。从实际数据看高热度不一定等于高评分有些低热度冷门片评分反而很高所以热度和口碑是两个独立维度不能简单混着看。这个结论也提醒我们在后续分析里一定要区分“叫好”和“卖座”。5. 进阶分析类型、演员和关键词基础指标看完了再把内容标签类数据利用起来。类型、演员、关键词这三类信息都能拆成一个个标签属于典型的“宽表转长表”分析也是 Spark 里 explode 最适用的场景。5.1 类型分布与类型组合TopN统计类型分布非常简单对 genre_names 数组列做 explode然后按类型名分组计数。val genreCounts base .filter($genre_names.isNotNull) .withColumn(genre, explode($genre_names)) .groupBy(genre) .agg(count(*).alias(movie_count)) .orderBy($movie_count.desc) genreCounts.show(15)但这里有一个细节一部电影可以属于多个类型所以 type 分布里同一部电影可能在 Drama 和 Thriller 里各计一次这不算 bug但解读结果时要清楚“计数”是类型出现次数不是电影数量。比单类型分布更有意思的是类型组合。很多电影不是只有一个类型而是几个类型同时存在比如“动作冒险”“剧情爱情”。如果直接拼接类型名顺序不同会导致组合被拆成两行。我的办法是先对数组做 sort_array 排序再拼接成 key这样“ActionDrama”和“DramaAction”会归到同一个组合。val genreCombos base .filter($genre_names.isNotNull) .groupBy(title) .agg(sort_array($genre_names).alias(genres_sorted)) .withColumn(combo, concat_ws(, $genres_sorted)) .groupBy(combo) .agg(count(*).alias(movie_count)) .orderBy($movie_count.desc) genreCombos.show(10)排序后再拼接这个思路在处理所有无序标签组合时都非常实用不只是电影类型。5.2 演员作品量和平均分explode后join演员信息在 credits 表的 cast 字段里结构比 genres 更复杂。这里需要定义 cast 的 JSON schema然后解析成数组再 explode 成演员维度。val castSchema ArrayType(StructType(Seq( StructField(cast_id, IntegerType), StructField(character, StringType), StructField(credit_id, StringType), StructField(gender, IntegerType), StructField(id, IntegerType), StructField(name, StringType), StructField(order, IntegerType) ))) val creditsParsed credits .withColumn(cast_arr, from_json($cast, castSchema)) .withColumn(actor, explode_outer($cast_arr)) .withColumn(actor_name, $actor.name) .select($movie_id.as(id), $actor_name)注意这里我用的是 explode_outer而不是 explode。区别在于如果某部电影的 cast 字段解析后是 nullexplode 会让这一行消失explode_outer 会把这一行保留下来actor_name 为 null。对演员分析来说丢几部其实影响不大但在生产环境里保留 null 往往更有价值因为你能统计出有多少数据是缺失的。接下来把演员表和宽表按电影 ID join然后按演员聚合val actorStats creditsParsed .join(base.select(id, vote_average, revenue), Seq(id), inner) .groupBy(actor_name) .agg( count(*).alias(movie_count), avg(vote_average).alias(avg_rating), sum(revenue).alias(total_revenue) ) .filter($movie_count 5) .orderBy($movie_count.desc, $avg_rating.desc) actorStats.show(15)join 之后还按 movie_count 过滤是为了排除只演过一两部电影的偶然性。要分析“稳定产出”的演员样本量太少没有意义。这个模块跑完既能找出作品数量最多的演员也能找出高口碑演员还能看出累计票房最高的演员是谁。5.3 关键词高频统计关键词字段在 movies.csv 的 keywords 列同样是一组 JSON 数组。解析方式和 genres 一模一样这也是为什么我在宽表阶段就把 keyword_names 建出来的原因。val keywordStats spark.read.parquet(output/tmdb_clean) .filter($keyword_names.isNotNull) .withColumn(kw, explode($keyword_names)) .groupBy(kw) .agg(count(*).alias(movie_count)) .orderBy($movie_count.desc) keywordStats.show(30)关键词统计可以反映某种内容主题的流行程度。比如“独立电影”“女性导演”“反乌托邦”这类标签出现频率越高说明这个主题在样本里的覆盖度越大。和类型组合一样关键词也属于无序标签如果有需要做组合分析同样要先 sort_array 再拼接。这套方法在标签系统、内容分类、用户画像场景里几乎可以原样复用。6. 开发期常见问题与调优实录最后一部分聊开发过程中真实遇到的坑。Spark 项目最耗时间的不是写逻辑而是处理数据格式、资源分配和性能问题。我把最有代表性的几个问题整理成排查记录给后来人做参考。6.1 CSV错位、日期空值和JSON解析失败第一个大坑就是 CSV 里带换行的长文本。如果没开 multiLineSpark 会把一条电影记录拆成多条导致 title、budget、genres 全部错位。排查方法很简单读进来之后先 count如果行数明显大于电影数量九成就是多行 CSV 没处理对。第二个坑是日期解析。release_date 列里有正常日期也有空字符串和脏格式。to_date 解析不认识的格式会直接返回 null不会报错。不少人看到 null 以为是函数有问题其实是被脏数据拖累了。我的处理方式是解析之后 filter 掉 null然后再提取年份。第三个坑是 from_json 解析失败静默返回 null。如果 genres_arr 大量为 null大概率是 schema 定义和数据不匹配。这时不要盲目改代码先拿一条原始 JSON 出来对照看字段名到底是 id/name还是其他名字。这个习惯能帮你省下好几个小时。6.2 数据倾斜怎么定位加盐两阶段聚合示例做演员聚合时容易遇到数据倾斜问题。比如一个热门演员出演了几百部电影按演员分组后包含这个演员的 task 要比其他 task 多处理很多数据导致整个 Stage 卡在那一个 task 上。定位方法很简单打开 Spark UI 看 Stage 页如果某个 task 的 Shuffle Read 或处理时间明显高于 others就需要处理倾斜。常用手段是加盐两阶段聚合。第一轮给 key 加一个随机后缀拆分大 key第二轮把后缀去掉再做一次聚合。val salted creditsParsed .withColumn(salt, (rand() * 20).cast(int)) .groupBy(actor_name, salt) .agg(count(*).alias(partial_cnt)) val result salted .groupBy(actor_name) .agg(sum(partial_cnt).alias(movie_count))需要注意如果还要计算平均分这类指标不能简单用这种 count 聚合。要做加权平均第一轮同时算出 count 和 avg第二轮用 sum(count * avg) / sum(count)。加盐不是越多越好盐值太大会引入额外的 shuffle 开销。小数据集和本机环境一般不会遇到明显倾斜但把方法掌握好生产环境里碰到大数据量就能直接套用。6.3 资源参数、落盘和Spark开发习惯开发期一定要调整 Spark 默认参数。除了上文说的本地 shuffle.partitions还要注意别在小数据集上滥用大配置。你有多少数据量就配多少资源否则只是增加调度开销。结果输出时不要一上来就 collect。collect 会把所有数据拉到 driver 端数据量一大直接 OOM。正确做法是先用 show 看几条确认无误后写到文件。如果输出文件很小可以用 coalesce(1) 合并成一个文件避免生成大量碎片文件。如果后续还要反复读取优先保存 Parquet不要保存 CSV。重复读 Parquet 比重新解析 CSV 快得多。另外Spark UI 是排查性能问题的第一现场。任务执行时间长先看是不是有数据倾斜提示内存不足先看是 driver 还是 executor再决定是调大内存还是优化 shuffle。不要一碰到 OOM 就盲目调参先看数据特征很多问题其实是代码写法造成的。把这套流程跑通之后再面对更大规模的数据无非是把 local 换成集群、把 CSV 换成分区表分析套路是通用的。写到最后分享一个我自己的习惯任何聚合结果先用 show 确认行数和采样再决定要不要落到磁盘千万别一上来就 collect 到本地这大概是 Spark 开发里性价比最高的一条经验。
