简介面向大数据课程期末项目这份压缩包提供了一个完整的电影推荐系统实现方案基于Spark、Hadoop HDFS与MongoDB技术栈使用Scala语言开发适合需要完成类似课程设计或入门分布式推荐系统的读者。包内共20个文件其中17个Scala源文件为核心代码覆盖数据预处理、特征工程、协同过滤与推荐生成等模块另有pom.xml、MANIFEST.MF与iml配置文件便于导入构建。压缩包整体仅18KB虽体积小巧但结构清晰包含完整的Maven工程组织。目前已有436人学习浏览适合作为大数据综合项目的参考模板。包内代码框架完整读者可理解Spark分布式计算、HDFS存储与MongoDB非关系型数据操作的整合方式同时体会Scala函数式编程在数据处理管道中的应用对掌握大数据生态核心组件具有直接的实践价值。1. 数据移动的速度决定了推荐结果的质量一次课程项目背后的架构直觉第一次打开某个大数据课程期末项目的代码目录时我的直觉是先找算法——电影推荐嘛协同过滤、矩阵分解这部分才是重头戏。可真把pom.xml和src/main翻完才发现真正难的不是那几十行 Scala 代码而是整个系统怎么在 Hadoop HDFS、Spark 和 MongoDB 之间搬运数据。分布式系统有一个朴素现实数据在哪里决定了计算会有多贵。同一份评分日志存在本地磁盘和存在 HDFS 上计算路径完全不是一个量级而训练出来的推荐结果存在 HDFS 还是 MongoDB又决定了业务方能不能用毫秒级查询拿到它。这个课程项目表面上是推荐系统实质上是一条数据管道。搞懂它为什么这样分层比调出一份好看的推荐列表更有价值。2. HDFS电影评分数据为什么要先切块复制而不是整文件读入内存2.1 存储层的数据布局块和副本是“并行读”的前提在课程项目里评分数据最初是 CSV。把它丢到 HDFS 后文件会被切成固定大小的块默认是 128MB每一块再保存多个副本。这里一个容易忽略的点是HDFS 并不只是“分布式存储”它同时把“读文件”变成了“并行读多个块”。推荐系统的训练阶段要反复扫描用户行为数据如果文件只放在一台机器上就算计算引擎换成 Spark也只有一个节点能读到数据其他节点都在空等。有了块和副本应用程序可以同时向多个 DataNode 发起读取每个 Executor 只处理其中一个块整体吞吐是单机读的若干倍。# 把本地评分数据放进 HDFS hdfs dfs -mkdir -p /movie_rec/input hdfs dfs -put ~/ml-latest/ratings.csv /movie_rec/input/ # 查看块和副本分布 hdfs fsck /movie_rec/input/ratings.csv -files -blocks -locationsfsck输出会列出块 ID 和每个副本所在的 DataNode是验证“数据到底存在哪”的最直接方式。常见的情况是CSV 只有几十 MB小于一个块大小于是 HDFS 上只形成一个块、三个副本。计算时虽然只有一个输入分区但副本让同一份数据可以同时被多个位置读取对容错的意义大于并行读。课程项目里经常会发现把 200MB 的评分文件放到三节点伪分布式集群上比放在本地磁盘上跑得更快原因不在磁盘速度而是 HDFS 把读取路径拆成了多块并行。副本数不是越多越好。三副本是 Hadoop 默认的取舍一个副本保证数据不丢第二个副本防止单机磁盘故障第三个副本配合机架感知策略防止整个机架断电。如果副本数提高到四或五写的网络开销会同步上升反而拖慢后续训练任务的输入速度。课程项目里没有必要调高副本数但有必要理解HDFS 的副本参数dfs.replication是在文件写入时生效的修改它只影响新写入的文件对已存在文件需要用hdfs dfs -setrep重新设置。2.2 NameNode 与 DataNode 的心跳节点失效后为什么还能继续训练NameNode 只保存目录、文件到块的映射关系真正的数据块分布在各 DataNode 上。DataNode 会周期性地向 NameNode 发送心跳与块报告证明自己还活着NameNode 一旦连续若干次收不到某个 DataNode 的心跳就会把该节点标记为宕机并把上面保存的副本标记为“副本数不足”。之后 NameNode 会启动复制任务在别处生成缺失的副本直到恢复三副本状态。这一过程对 Spark 计算层是透明的训练作业读到的始终是逻辑文件路径物理块的位置由 HDFS 调度。这个设计直接决定了推荐训练的稳定性。模型训练动辄十几分钟如果某个节点中途宕机依赖 HDFS 这块“数据保险”Spark 可以重新在其余节点上调度任务而不是让整个fit过程从头再来。HDFS 默认的副本放置策略是第一个副本放在客户端所在的节点第二个副本放在另一个机架的某个节点第三个副本放在第二个副本所在机架的另一个节点。这样设计是为了同时照顾写入速度和可用性本地写第一副本不经过网络跨机架的第二副本防止机架级故障第二、第三副本同机架则减少了恢复副本时的跨机架流量。我一般会这样理解HDFS 用“空间换时间”的思路用额外副本换来了“读并行度”和“故障自动转移”。课程项目里看到的三份数据副本看似浪费了磁盘实际上是把推荐系统最担心的“运行到一半数据丢了”变成了一个小概率事件。对新手来说最容易踩的误区是以为 HDFS 是“把文件分配到多台机器存储”其实它更准确的语义是“把文件的块分配到多台机器并冗余存储”。2.3 原始数据放在 HDFS中间结果放哪里目录与格式也是设计课程项目的数据流往往是这样原始评分 CSV 进入 HDFS经过 Spark 预处理后生成特征表训练完成后输出推荐结果。很多人只关注算法代码忽略了 HDFS 目录结构本身就是可维护性的一部分。一个清晰的布局通常长这样/movie_rec/raw/ratings.csv /movie_rec/raw/movies.csv /movie_rec/feature/user_stats.parquet /movie_rec/als_model/ /movie_rec/output/top_rec_per_user.parquet这样的目录设计把“未加工的原始数据”和“经过清洗的特征数据”分开。原因是原始数据只写一次后续无论怎么修改特征工程都不应该覆盖它特征数据则是某个计算流程的产物可以随时重建。如果每一次训练都从/movie_rec/raw开始虽然逻辑最简单但重复读取和解析 CSV 会把大量时间浪费在整行扫描上。生产环境通常会把手动put进来的 CSV 转成 Parquet 列式存储因为特征读取经常只取 sample 若干列列存能大幅减少 IO。HDFS 本身对随机写支持得不好追加写也有较多限制。这反而让推荐系统的批处理模式变得很自然每个阶段产生一份完整的新数据下一个阶段整体读取上一阶段的输出阶段之间不试图原地更新。理解了这一点就不会问“为什么 Spark 运行完还要把结果重新写一遍 HDFS”这种问题。课程项目里HDFS 承担的是“数据仓库”角色所有上游产生的中间数据在这里落地给计算引擎一个稳定、可重放的数据源。3. SparkRDD 分区、DAG 调度和 Shuffle 如何把一次训练拆成接力赛3.1 分区是并行边界内存是第二存储Spark 程序里数据集被抽象为 RDD 或 DataFrame。一个输入文件有几个 HDFS 块默认就会形成几个分区每个分区由 Executor 里的一个 task 处理。分区数决定了并行度如果集群有 16 个 Executor但输入只有一个分区那么 15 个 Executor 会空闲作业跑得跟单机一样慢。反之如果分区数达到几千每个 task 只处理几 KB 数据调度开销反而会超过计算开销。课程项目里最常见的性能问题不是算法不够好而是分区数设置和集群规模、数据量根本不匹配。Spark 比第一代 Hadoop MapReduce 强在“迭代中间结果放内存”。ALS 这样的推荐算法需要反复迭代每一轮都要使用上一轮产出的用户因子矩阵和电影因子矩阵。如果每个迭代轮次都把矩阵写回 HDFS下一轮再读出来单次迭代的网络和磁盘开销就会把算法效率拖垮。Spark 的做法是DAG 调度器把整个作业切成多个 stage每个 stage 内部尽量在内存中完成计算只有遇到宽依赖才把中间结果落盘。这个设计是“内存计算”的真正含义并不是所有数据都常驻内存而是把频繁复用的迭代数据放在内存里。RDD 血缘lineage是另一层保障。每个 RDD 都记录了自己是从哪个父 RDD 经过什么转换得来的。一旦某个分区的计算结果丢失Spark 不需要重新跑整个作业而是根据血缘关系只重新计算丢失的那一个分区。课程项目里如果看到某个 ALS 训练任务失败后重试只花了几秒就说明血缘机制只补了少量数据而非把几十个 stage 重新执行一遍。3.2 窄依赖与宽依赖以及 Shuffle 的代价RDD 依赖分为窄依赖和宽依赖。窄依赖指父 RDD 的每个分区最多被子 RDD 的一个分区使用例如map、filter这类转换可以在同一个节点内完成不需要移动数据。宽依赖指父 RDD 的每个分区可能被多个子分区使用典型操作是groupByKey、reduceByKey、distinct和未优化过的join。宽依赖会触发 Shuffle把数据按照新的 key 重新分区通过网络传输到对应的 Executor。Shuffle 是 Spark 作业里最贵的操作因为它伴随序列化、磁盘写和网络传输。电影推荐任务里有两处典型宽依赖。第一处是“统计每个用户的评分数量”必须把同一 userId 的评分记录从不同分区汇聚到一起这个groupBy必然产生 Shuffle。第二处是“把评分数据和电影元数据做 join 得到电影标题”如果两张表都很大join 通常也要 Shuffle。理解宽依赖后优化方向就清晰了尽量在 Shuffle 之前缩小数据量。先过滤掉无效评分再按用户聚合先选出推荐候选需要的电影字段再与元数据 join。常见的原则是“先窄后宽先把数据变瘦再让数据搬家”。课程项目里还有个容易踩的细节join不一定会触发全量 Shuffle。如果其中一张表很小Spark 会把小表广播到每个 Executor直接在本地做 hash join避免了 shuffle。这也是为什么建议把电影元数据这种几十 MB 的小表用broadcast广播而不要让它和几千万行评分日志一起走整网络传输。理解窄依赖和宽依赖之后看 Spark UI 上的 stage 划分会非常直观一个 stage 结束后必然出现 shuffle 边界时间长、任务多的 stage 往往是宽依赖之后的计算。3.3 Scala 管道代码从原始评分到 ALS 输入课程项目的源码包src/main下核心动作可以浓缩成下面这样一段 Scala 代码。它不是项目里的完整实现但代表了我见过的大多数课程作业会采用的骨架// 读 HDFS 上的原始评分把字符串类型转成数值类型 import org.apache.spark.ml.feature.StringIndexer import org.apache.spark.ml.recommendation.ALS val raw spark.read .option(header, true) .csv(hdfs:///movie_rec/raw/ratings.csv) // 过滤掉缺失和异常评分避免后面 ALS 直接报错 val ratings raw .selectExpr( cast(userId as int) as userId, cast(movieId as int) as movieId, cast(rating as double) as rating ) .filter(rating is not null and rating between 0.5 and 5.0) // userId、movieId 可能是字符串需要离散化成从 0 开始的连续整数 val userIndexer new StringIndexer() .setInputCol(userId) .setOutputCol(userIdx) val itemIndexer new StringIndexer() .setInputCol(movieId) .setOutputCol(movieIdx) val indexed itemIndexer.fit(ratings) .transform(ratings) .transform(userIndexer.fit(ratings))这段代码里最有价值的逻辑是StringIndexer。ALS 的内部实现会把用户 ID、电影 ID 当作矩阵下标如果原始 ID 不是连续的整数因子矩阵会变成稀疏矩阵训练内存和耗时都会爆炸。StringIndexer把离散字符串映射到连续整数是几乎所有基于 Spark MLlib 的推荐项目都绕不开的预处理步骤。接着才是训练动作val als new ALS() .setRank(20) // 隐因子维度越大拟合能力越强但参数也越多 .setMaxIter(10) // 迭代轮数课程项目 10 次左右足够观察收敛 .setRegParam(0.1) // 正则化系数防止隐因子过拟合 .setUserCol(userIdx) .setItemCol(movieIdx) .setRatingCol(rating) .setColdStartStrategy(drop) // 测试集中新用户/新电影直接丢弃避免产生 NaN val model als.fit(indexed)参数rank、maxIter、regParam是三个最值得调的值。rank决定隐因子向量的长度课程数据集通常 10 到 30 足够regParam太大会让模型偏向平均值太小则几乎不过拟合。setColdStartStrategy(drop)是必须留意的一项训练集里没出现过的用户或电影在预测阶段没有对应因子向量不处理会输出 NaN导致下游写 MongoDB 时直接数据异常。4. MongoDB训练结果为什么要落到文档数据库而不是直接留在 HDFS4.1 ALS 训练出来的两个矩阵如何变成 Top N 推荐协作过滤算法并不直接预测所有用户对全部电影的打分那样做的计算量是用户数乘电影数课程项目里也完全没必要。ALS 把评分矩阵分解成用户因子矩阵和电影因子矩阵两个矩阵相乘就能得到任意用户对任意电影的预测评分。课程项目通常只关心“每个用户最喜欢哪些电影”所以训练完成后不会把完整预测矩阵输出而是为每个用户计算出 Top N 候选列表。这一步在 Spark 上常见的做法是对每个用户做model.recommendForAllUsers(20)得到的结果是结构化的 DataFrame每一行对应一个用户和该用户预测分最高的 20 部电影。要把它写入 MongoDB最直接的方式是输出为嵌套文档而不是平铺成 20 行。嵌套结构文档在 MongoDB 里是天然合法的{ userId: 837, recommendations: [ { movieId: 296, score: 4.72 }, { movieId: 112, score: 4.61 }, { movieId: 587, score: 4.55 } ], generatedAt: 2024-06-01T12:00:00Z }这个 JSON 文档的含义是“某用户一次推荐的结果快照”。每条recommendations子文档保存电影 ID 和预测分。在 MongoDB 里对userId建唯一索引后查询某个用户最近一次推荐结果只需要一次索引点查耗时通常在毫秒级。这样的数据模型比把结果写成 CSV 再放回 HDFS更适合“按用户查询”的业务场景。4.2 为什么是 MongoDB而不是 HDFS 或关系型数据库HDFS 的设计目标是全量扫描随机读取某一个用户的推荐结果非常低效哪怕用hdfs dfs -cat去过滤也要把整个文件扫完。关系型数据库适合结构化数据但课程项目里推荐列表是嵌套数组用一张关联表和多次 join 才能还原出来。MongoDB 的文档模型刚好避开了这两者的麻烦一张表对应一个集合一个用户一条文档推荐列表直接作为子数组存储。三种存储的选择差异可以用一张表说明存储方案读模式写入模型适合推荐系统的角色HDFS全量扫描追加、整体覆盖原始日志、批量中间结果关系型数据库索引点查事务性更新电影、用户等强结构元数据MongoDB索引点查、聚合文档替换推荐结果、用户画像等半结构化数据课程项目里常见的分工是HDFS 存不可变的原始日志MongoDB 存清洗后的电影详情、用户画像和最终推荐结果。有人会问为什么不用 RedisRedis 是内存数据库适合存高热数据的短时缓存但对课程项目而言推荐结果需要能跨次运行保存、能被业务系统反复按用户 ID 查询MongoDB 的持久化能力和文档结构更合适。4.3 Spark 写 MongoDB 的代码实现Spark 官方和 MongoDB 社区都提供 connector把 DataFrame 直接写进 MongoDB 集合。课程项目里常见的写法是这样import com.mongodb.spark.sql._ recommendationsDF.write .format(mongo) .mode(overwrite) .option(database, movie_rec) .option(collection, recommendations) .save()这段代码的作用是把 Spark 计算结果批量写入 MongoDB。mode(overwrite)表示整集合覆盖适合离线批处理场景模型每天或每周重新训练一次训练完把整张推荐结果替换掉。写入时 Spark 会按分区并发执行每个 executor 负责一部分用户的写入不需要像传统程序那样一条条插入。在线查询端只需要几行代码就能拿到推荐列表db.recommendations.findOne({userId: 837})。这正是课程项目把 MongoDB 放在数据链路末端的原因计算层的输出是面向全量用户的批量文件服务层需要的是面向单个用户的点查。HDFS 擅长前者MongoDB 擅长后者两者之间通过 Spark connector 打通。5. 避坑Spark HDFS MongoDB 这条链路里最常翻车的四个位置5.1 小文件灾难几十个分区写出来的结果是散弹现象训练完成hdfs dfs -ls /movie_rec/output看到目录下有几百个文件其中大部分只有几 KB甚至空文件。下一次作业读取这个目录时明显变慢NameNode 内存占用也直线上升。原因Spark 写文件时每个 task 会自己写出一个文件。并行度设成了 500最终就会生成 500 个文件不管每个文件多大。大量小文件在 HDFS 上意味着大量元数据和大量小的读请求严重时会让 NameNode 成为瓶颈。解决在写结果之前主动降低分区数。df.coalesce(20).write.parquet(...)或df.repartition(20).write...把输出文件数量控制在一个与后续读取并发度匹配的水平。课程项目的结果数据通常很小合并成几个文件完全没有问题。关键是理解分布式计算里的并行度是计算时的概念落地存储时文件数要按存储系统的脾气来。5.2 数据倾斜少数“影迷用户”拖慢整个训练现象ALS 或 groupBy 阶段大多数 task 几十秒就完成个别 task 跑了十分钟还没结束Spark UI 上某个 stage 的进度条长期卡在 99%。原因评分数据不是均匀分布的。少数用户看了几千甚至上万部电影按 userId 分组时这些用户的评分记录全部落在一个或几个分区里负责这些分区的 task 就成了拖后腿的瓶颈。解决在聚合类操作之前先观察数据分布用groupBy(userId).count().describe()排查。对于少数超大 key可以在 groupBy 之前给 userId 加一个随机后缀把一个大 key 拆成多个子 key分别聚合后再合并。课程项目不需要非常复杂的倾斜处理但要能识别出倾斜的典型现象而不是盲目调大 executor 内存。内存调大了只会推迟故障并不会减少需要处理的数据量。5.3 MongoDB 连接数被打满现象Spark 作业写 MongoDB 时MongoDB 日志出现大量 rejected connection应用侧报 MongoTimeoutException 或 connection 无法建立。原因Spark 的 executor 默认并发度很高几十个 task 同时写入 MongoDB 时各自的连接池叠加起来会迅速占满 MongoDB 的可用连接数。MongoDB 对每个连接的内存开销并不低连接数过多会直接压垮服务。解决写 MongoDB 之前先coalesce(10)或coalesce(5)大幅降低写入并发度。同时检查spark.mongodb.write.connection.connectionPoolSize之类的连接池配置把每个 executor 的连接数调小。更简单的办法是不要用overwrite反复整表覆盖而是先把新结果写成临时集合确认成功后用 rename 替换减少写入期间对旧查询的影响。5.4 Scala 隐式转换与类型擦除编译过了运行期还是崩现象代码编译没有报错但运行时抛ClassCastException例如[Ljava.lang.Object; cannot be cast to ...。这个问题通常在处理 RDD 与 DataFrame 互转或使用 Scala 集合与 Java 集合互操作时出现。原因Scala 的隐式转换在编译期插入类型转换但 JVM 的类型擦除让泛型在运行时只剩下原始类型。Spark 的map算子返回的对象经过序列化、洗牌后类型信息并没有像编译期那样严格一旦写代码时依赖了隐式转换生成的类型运行期就可能对不上。解决少依赖隐式转换多写显式类型标注。从 DataFrame 取数据列时用getAs[Int](userId)而不是asInstanceOf[Int]定义 Scala 集合泛型时写val list: List[Int]不要只写List。项目里如果某段代码要从 ALS 的中间结果提取向量优先用org.apache.spark.ml.linalg.Vector的现成方法不要自己写循环做asInstanceOf。这类问题最难以排查但如果能形成“所有转换必须显式写类型”的习惯大部分 ClassCastException 在代码审查阶段就能避免。6. 取舍批处理先于实时课程项目真正教会你的是什么事情课程项目选择了一个离线批处理架构HDFS 存原始数据Spark 定期训练模型MongoDB 存放结果并对外查询。为什么不直接上实时推荐因为实时的前提是离线链路已经稳定。生产环境里冷启动、用户实时点击、特征更新都会让实时推荐复杂很多而课程项目能证明的是“数据从 HDFS 进、经过 Spark 算、落到 MongoDB 出”这件事可以闭环。有了这个闭环后续换成实时增量训练只是替换入口和触发方式核心的存储模型和数据流不会变。训练完模型之后最值得做的验证不是肉眼看推荐列表顺不顺眼而是把评分数据切出一份测试集计算预测分与真实分的误差。常见指标是 RMSE如果模型只记住了训练集里的评分而没学会规律测试集 RMSE 会明显偏高。课程项目里可以按时间戳把后 20% 的评分作为测试集训练集和测试集按用户隔离避免同一个用户既出现在训练集又出现在测试集里造成营业错觉。从那以后我每次拿到一个数据处理任务都会强制自己先画一张“数据从哪来、在哪算、到哪去”的草图标清楚每一步的读模式是扫描还是点查再开始写代码。很多看似是算法或调参的问题最后都出在数据移动的方式上。如果这份课程项目的源码和文档能帮你建立起同样的直觉那它带给你的价值就不只是一门课的分数。希望帮到你。本文还有配套的精品资源点击获取