MongoDB + Spark 大数据项目实战：从环境搭建到数据清洗聚合的完整链路-北京尧图网络

简介本资源为基于MongoDB与Spark的大数据项目完整资料包面向计算机、人工智能、通信工程、自动化等专业的在校学生、教师及企业员工可用于毕业设计、课程设计、作业提交或项目初期立项演示也适合具备一定基础的小白进阶学习。压缩包共81个文件约375KB以60个Java源文件为核心实现业务逻辑辅以6个JSP页面、4个XML配置、2个JS脚本及CSS样式另含README说明文档与项目截图整体结构清晰、便于按模块查阅。该项目为个人高分作品已通过导师指导与答辩评审获得95分评价代码均经过测试运行成功后才上传功能完整可靠。读者可从中获取完整源码、项目文档与优秀项目参考理解MongoDB与Spark的整合思路、目录组织方式及关键实现细节并在此基础上修改扩展实现其他功能或直接用于毕设、课设等场景。目前已有60人学习关注欢迎下载交流、共同进步。1. 从一份 MongoDB Spark 资料包说起它到底能解决什么问题很多人第一次接触“基于 MongoDB Spark 的大数据项目”时手里拿到的往往是一个压缩包里面塞着文档、源码和一堆配置样例。真正让人卡住的不是“有没有资料”而是不知道这套组合拳该从哪里下手MongoDB 负责存什么Spark 负责算什么两者之间的数据通道怎么打通跑起来之后又该看哪些指标判断它是不是真的在工作。这个标题背后对应的其实是一条非常典型的落地链路——用文档型数据库承接原始或半结构化的业务数据再用 Spark 做分布式清洗、聚合和分析最后把结果写回库或者导出给下游。它适合两类人一类是正在做课程设计或毕业设计、需要一套能跑通的最小闭环的学生另一类是已经会写单机脚本、但没真正把 Spark 和 MongoDB 接起来跑过数据流的工程师。下面我按自己实际搭环境的顺序把这条链路拆开讲清楚。2. 环境准备MongoDB 与 Spark 的安装和连通性验证2.1 为什么先装 MongoDB 而不是先配 Spark我见过不少人一上来就折腾 Spark 集群结果 Spark 跑通了数据源连不上回头再查 MongoDB 的认证和网络配置时间全浪费在来回切换上。合理的顺序是先把 MongoDB 单节点跑起来确认能写入、能查询再装 Spark最后做两者之间的连接测试。MongoDB 的安装本身不复杂但“安装失败”是热搜里高频出现的问题绝大多数不是包的问题而是系统依赖、数据目录权限和配置文件三件事没对齐。在 Linux 上常见做法是用官方提供的包管理方式安装安装完成后不要急着改一堆参数先用默认配置启动确认服务能起来。下面这段命令是我一般在 Ubuntu 或 Debian 系机器上用的流程关键点在于数据目录和日志目录必须提前建好并授权给 mongodb 用户否则服务启动会直接报权限错误。# 导入 MongoDB 公钥并添加软件源以常见 LTS 版本为例 wget -qO - https://www.mongodb.org/static/pgp/server-7.0.asc | sudo apt-key add - echo deb [ archamd64,arm64 ] https://repo.mongodb.org/apt/ubuntu jammy/mongodb-org/7.0 multiverse | sudo tee /etc/apt/sources.list.d/mongodb-org-7.0.list # 更新并安装 sudo apt-get update sudo apt-get install -y mongodb-org # 创建数据与日志目录并授权给 mongodb 用户 sudo mkdir -p /var/lib/mongodb /var/log/mongodb sudo chown -R mongodb:mongodb /var/lib/mongodb /var/log/mongodb # 启动服务并设为开机自启 sudo systemctl start mongod sudo systemctl enable mongod # 验证服务状态 sudo systemctl status mongod这段脚本里真正容易翻车的是目录权限那两行。很多人用 root 直接跑 mongod短期能起来但一旦用 systemctl 托管就会因为用户不匹配而失败。参数上/var/lib/mongodb是数据文件落盘位置/var/log/mongodb是日志位置这两个路径要和/etc/mongod.conf里的storage.dbPath和systemLog.path保持一致不一致时服务会按配置文件为准你建的目录就白建了。装完之后不要只满足于systemctl status显示 active要真正连进去做一次写入和查询。这一步是后面 Spark 读取的前提。# 进入 MongoDB shell mongosh # 切换到测试库并写入一条文档 use bigdata_demo db.user_events.insertOne({ user_id: u1001, event_type: click, item_id: item_88, ts: new Date(), province: Zhejiang }) # 查询确认写入成功 db.user_events.find({ user_id: u1001 }).pretty() # 查看集合列表 show collections这里涉及 MongoDB 的三个核心概念数据库database对应bigdata_demo集合collection对应user_events文档document就是那条带user_id、event_type的 JSON 结构记录。Spark 后续读取时就是按“数据库.集合”这个粒度去拉数据的。写入成功后建议再执行一次删除操作确认权限没问题因为 Spark 在写入结果时同样需要写权限。# 删除刚才那条测试文档验证删除权限 db.user_events.deleteOne({ user_id: u1001 }) # 确认删除结果 db.user_events.countDocuments({ user_id: u1001 })2.2 Spark 安装与读取 MongoDB 的依赖配置Spark 本身不绑定 MongoDB真正让两者对话的是 MongoDB 提供的 Spark Connector。很多人 Spark 装好了代码也写了一跑就报ClassNotFoundException或者NoSuchMethodError根因基本都是连接器版本和 Spark 版本没对上。我的习惯是先确定 Spark 版本再去选对应的 Connector而不是反过来。Spark 的安装可以走本地模式先跑通不必一上来就搭集群。本地模式对验证逻辑足够用等逻辑跑通再迁到集群能省掉大量排查网络和资源调度的时间。# 下载并解压 Spark以 3.5.x 预编译版为例具体版本按你环境选 wget https://archive.apache.org/dist/spark/spark-3.5.1/spark-3.5.1-bin-hadoop3.tgz tar -zxvf spark-3.5.1-bin-hadoop3.tgz -C /opt/ mv /opt/spark-3.5.1-bin-hadoop3 /opt/spark # 配置环境变量 echo export SPARK_HOME/opt/spark ~/.bashrc echo export PATH$SPARK_HOME/bin:$PATH ~/.bashrc source ~/.bashrc # 验证 Spark 可用 spark-shell --version版本确认之后写一个最小的 PySpark 脚本去读 MongoDB。这里的关键是spark.jars.packages参数它会在启动时自动拉取 Connector 的 jar 包。参数格式是org.mongodb.spark:mongo-spark-connector_2.12:10.3.0其中2.12是 Scala 版本要和你的 Spark 预编译版本一致10.3.0是 Connector 版本不同大版本之间 API 有差异不要混用。# mongo_spark_read.py from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(MongoSparkReadDemo) \ .config(spark.mongodb.read.connection.uri, mongodb://127.0.0.1:27017/bigdata_demo.user_events) \ .config(spark.mongodb.write.connection.uri, mongodb://127.0.0.1:27017/bigdata_demo.result_events) \ .config(spark.jars.packages, org.mongodb.spark:mongo-spark-connector_2.12:10.3.0) \ .getOrCreate() # 读取集合 df spark.read.format(mongodb).load() df.printSchema() df.show(5, truncateFalse) spark.stop()这段代码里spark.mongodb.read.connection.uri指定读哪个库的哪个集合write那条指定写回哪里。printSchema用来确认字段类型有没有被正确推断尤其是时间字段MongoDB 里是 Date 类型Spark 侧可能映射成 Timestamp 或 String这会影响后续聚合。show用来肉眼确认数据确实拉过来了。如果这一步报连接超时先检查 MongoDB 是否监听在127.0.0.1以及是否开启了认证如果报类找不到就是 Connector 版本和 Spark 不匹配。3. 用 Spark 做数据清洗与聚合从原始文档到可用结果3.1 读取 JSON 风格文档后的字段规整MongoDB 里的文档天然是嵌套结构Spark 读进来之后嵌套字段会变成 StructType直接做聚合会很别扭。我一般会先把需要的字段拍平再做过滤和分组。这一步对应热搜里“spark中读取json”和“基于spark的数据清洗”这两个高频需求本质是同一件事把半结构化数据整理成规整的表结构。假设user_events集合里每条文档除了顶层字段还有一个device子对象包含os和version。下面这段代码演示怎么把子字段取出来变成独立列同时过滤掉明显无效的记录。from pyspark.sql import SparkSession from pyspark.sql.functions import col, to_timestamp, when spark SparkSession.builder \ .appName(MongoSparkClean) \ .config(spark.mongodb.read.connection.uri, mongodb://127.0.0.1:27017/bigdata_demo.user_events) \ .config(spark.jars.packages, org.mongodb.spark:mongo-spark-connector_2.12:10.3.0) \ .getOrCreate() raw_df spark.read.format(mongodb).load() # 拍平嵌套字段统一时间格式过滤无效 user_id clean_df raw_df.select( col(user_id), col(event_type), col(item_id), col(province), col(device.os).alias(device_os), col(device.version).alias(device_version), to_timestamp(col(ts)).alias(event_time) ).filter( col(user_id).isNotNull() (col(user_id) ! ) ) clean_df.printSchema() clean_df.show(5, truncateFalse)col(device.os)这种写法是取嵌套字段的标准方式比先 explode 再拼回去高效得多。to_timestamp把时间统一成 Spark 的 Timestamp 类型后面做时间窗口聚合才不会出错。过滤条件里同时判空和判空字符串是因为有些脏数据会写成空串而不是 null只判 null 会漏掉。这一步做完数据就从“文档”变成了“表”后续所有 Spark SQL 和 DataFrame 操作都能直接套用。3.2 分组聚合与结果写回 MongoDB清洗完的数据要产生价值通常得做聚合。比如按省份和事件类型统计点击量这是最典型的 Spark 数据分析案例。聚合之后的结果可以写回 MongoDB 的一个新集合也可以导出成其他格式。写回时要注意写入模式append是追加overwrite是覆盖生产环境里我一般用append配合分区字段避免误删历史数据。# 按省份和事件类型聚合 agg_df clean_df.groupBy(province, event_type).count() \ .withColumnRenamed(count, event_count) agg_df.show(10, truncateFalse) # 写回 MongoDB 新集合 agg_df.write \ .format(mongodb) \ .mode(append) \ .option(spark.mongodb.write.connection.uri, mongodb://127.0.0.1:27017/bigdata_demo.result_events) \ .save() spark.stop()groupBy后面跟count()是最基础的聚合实际项目里还会叠加window做时间窗口、叠加approx_count_distinct做去重计数。写回时mode(append)表示不覆盖已有数据如果你在调试阶段反复跑建议先用overwrite清掉旧结果确认逻辑无误后再换成append。写回完成后回到 mongosh 里查一下result_events集合确认结果条数和预期一致这一步是验证整条链路是否闭环的关键。mongosh use bigdata_demo db.result_events.find().pretty() db.result_events.countDocuments()如果result_events里没有数据先看 Spark 任务日志里有没有写入异常再确认 MongoDB 用户对目标集合是否有写权限。很多时候不是代码问题而是认证库和业务库不是同一个导致写入被拒。4. 避坑与排查MongoDB Spark 联调中最容易翻车的 5 个点4.1 连接器版本不匹配导致任务直接失败现象是 Spark 任务一启动就抛NoClassDefFoundError或NoSuchMethodError堆栈里出现mongo-spark-connector相关类名。原因几乎都是 Connector 的 Scala 版本和 Spark 预编译版本不一致比如 Spark 是 Scala 2.12 编译的你却拉了_2.13的包。解决办法是先执行spark-shell --version确认 Scala 版本再把spark.jars.packages里的后缀改成对应版本并且 Connector 大版本不要跨代混用。4.2 MongoDB 认证配置与连接串不匹配现象是本地 mongosh 能连能查但 Spark 一读就报Authentication failed或Command failed with error 13。原因是 MongoDB 开启了认证但连接串里没带认证库或者用户是建在admin库却试图在业务库里认证。解决办法是在连接串里显式指定authSource例如mongodb://user:pass127.0.0.1:27017/bigdata_demo?authSourceadmin并确认该用户对目标库有 readWrite 权限。4.3 数据倾斜导致聚合阶段个别任务卡死现象是 Spark UI 里大部分 task 很快完成但少数几个 task 长时间不动整体进度卡在 99%。原因是某个province或event_type的数据量远超其他值导致单个分区压力过大。解决办法是先对聚合键做采样确认是否存在超级热点再考虑加盐打散或者两阶段聚合。我一般会先用groupBy加count排序看一眼分布再决定要不要动分区策略。4.4 时间字段类型不一致导致窗口聚合报错现象是代码里用window做时间窗口时抛类型异常提示字段不是 timestamp。原因是 MongoDB 里ts字段有的文档存的是字符串有的存的是 DateSpark 推断出来是 String 类型。解决办法是在清洗阶段强制用to_timestamp转换并对转换失败的记录做标记或过滤不要指望 Spark 自动帮你兜底。4.5 写回模式选错导致历史结果被清空现象是每次跑任务result_events集合里的数据都只剩最新一批历史结果全没了。原因是写入时用了mode(overwrite)而调试阶段没注意。解决办法是调试期用overwrite没问题但上线前一定改成append并且最好在写入前对目标集合做一次备份或者用带时间戳的集合名做版本隔离。5. 进阶技巧用分区裁剪和写入优化把任务跑得更稳5.1 读取时下推过滤条件减少数据传输Spark 读 MongoDB 时如果能把过滤条件下推到数据库侧就能大幅减少网络传输和内存占用。Connector 支持通过aggregation.pipeline参数传入 MongoDB 的聚合管道把$match阶段放在数据库端执行。下面这段配置只读取event_type为click且时间在指定范围内的文档比全量拉过来再 filter 高效得多。spark SparkSession.builder \ .appName(MongoSparkPushdown) \ .config(spark.mongodb.read.connection.uri, mongodb://127.0.0.1:27017/bigdata_demo.user_events) \ .config(spark.mongodb.read.aggregation.pipeline, {$match: {event_type: click}}) \ .config(spark.jars.packages, org.mongodb.spark:mongo-spark-connector_2.12:10.3.0) \ .getOrCreate() df spark.read.format(mongodb).load() df.show(5)aggregation.pipeline的值是一个 JSON 字符串里面写 MongoDB 原生的聚合阶段。$match是最常用的下推阶段$project也可以用来只保留需要的字段。注意管道里的字段名要和 MongoDB 文档里的实际字段一致写错了不会报错只会返回空结果这点很容易让人误以为连接断了。5.2 写入时控制分区数和批量大小写回 MongoDB 时如果 Spark 分区数太多会建立大量并发连接反而拖慢写入甚至打爆数据库连接池。我一般会在写入前用coalesce把分区数压到一个合理范围比如 4 到 8 个具体看 MongoDB 实例的承载能力。同时可以调整批量写入大小减少往返次数。# 控制写入分区数避免连接过多 agg_df.coalesce(4).write \ .format(mongodb) \ .mode(append) \ .option(spark.mongodb.write.connection.uri, mongodb://127.0.0.1:27017/bigdata_demo.result_events) \ .option(spark.mongodb.write.batchSize, 1000) \ .save()coalesce(4)把分区数降到 4batchSize设为 1000 表示每批写入 1000 条文档。这两个参数没有万能值要根据数据量和数据库规格调。我的习惯是先按默认跑一次看 Spark UI 里写入阶段的耗时和 MongoDB 的监控指标再决定是调大还是调小。调完之后一定要重新跑一次全量验证确认结果条数和聚合值没有因为分区变化而改变。5.3 用计数校验代替肉眼抽查最后分享一个我踩过坑之后养成的习惯每次任务跑完不要只看show出来的几行数据就认为对了一定要做计数校验。比如清洗前后的记录数差值是否等于被过滤掉的脏数据量聚合结果的 sum 是否和原始数据里对应字段的 sum 一致。这些校验用几行 Spark SQL 就能完成但能帮你挡住绝大多数“看起来跑通了、其实数据错了”的情况。# 校验清洗后记录数被过滤记录数原始记录数 raw_count raw_df.count() clean_count clean_df.count() filtered_count raw_count - clean_count print(fraw{raw_count}, clean{clean_count}, filtered{filtered_count}) # 校验聚合结果的总和是否等于清洗后数据的总条数 agg_total agg_df.agg({event_count: sum}).collect()[0][0] print(fagg_total{agg_total}, clean_count{clean_count}) assert agg_total clean_count, 聚合总数与清洗后记录数不一致检查分组逻辑这段校验代码里agg_total应该严格等于clean_count如果不等说明分组或过滤逻辑有问题。断言失败时任务会直接抛异常比事后人工核对靠谱得多。这套习惯让我在多个项目里提前发现了字段映射错误和重复计数问题希望帮到你。本文还有配套的精品资源点击获取

MongoDB + Spark 大数据项目实战：从环境搭建到数据清洗聚合的完整链路

相关推荐

Spark电商用户画像实战：数据管道、特征工程与调优全解析

基于Matlab与Simulink的10机39节点电力系统仿真建模与调试

Python数据挖掘实践：k-means聚类与二手房数据分析全流程

更多资讯

AI数据合规全链路指南：从训练语料到模型输出的落地实践

高效演讲实操手册：建立听众思维，让说话真正惹人爱

2026 办公 Agent 工具选型手册：5 大类 12 款横向对比表（附快速选型）与 TaoToken 统一接入实践

Python与NumPy从零实现BP神经网络手写字体识别源码详解

Java复用进阶：final关键字、初始化顺序与类加载机制全解析

open-code-review：结构化代码评审工具的设计与落地实践

今日更新

Susi AI API实战指南：susi_alexa_skill如何用Node.js调用chat.json获取智能回答

换新电脑了？KeyStats 恢复码数据找回完全指南，端到端加密统计一键重建

重建二叉树：由前序与中序遍历序列还原二叉树的原理与 Java 实现（YCBlogs 剑指 Offer 精讲）

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？