Spring Boot与Spark构建共享单车数据存储与聚合系统

简介：针对SpringBoot与Spark结合开发共享单车数据存储系统的毕业设计项目，完整包含后端Java源码、前端Vue页面、论文文档及数据库脚本。项目以共享单车使用数据为场景，演示了从数据采集、分布式存储到Spark分析处理、SpringBoot接口交付的完整链路，适合计算机科学与技术、人工智能等专业学生用于毕业设计或课程实践。压缩包内共374个文件，以java、vue、js、svg、xml、sql、doc/docx等类型为主，涵盖前后端工程代码、界面图标、配置文件、数据库初始化脚本、论文与操作文档以及bat启动脚本，整体约17.8MB，结构清晰便于按模块查阅。目前已有51人学习下载。通过这套资料，可以快速搭建运行环境，重点理解SpringBoot如何集成Spark进行大数据分析，同时参考论文写作思路，为同类课题提供落地范本。

1. 从毕设标题到现实数据链路：共享单车数据存储系统到底解决什么问题

最初接触到“springboot基于Spark的共享单车数据存储系统”这个方向时，我并没有急着搭环境，而是先估算了一下数据压力：一座二线城市的中型共享单车运营商，高峰期每天能产生 40 ～ 100 万条骑行订单，每条订单包含车辆编号、用户编号、起止时间、起止经纬度、骑行时长、计费金额等十几个字段。若按传统思路用 Spring Boot 的 service 层逐行解析 CSV、再循环 INSERT 进 MySQL，数据库连接池会先被打满，接口响应也会被长时间占住。这套系统的设计核心，是把“数据管道”和“业务接口”拆成两层：Spark 负责对原始骑行记录做清洗、去重、聚合，并把统计结果写入 MySQL；Spring Boot 只对已经成型的统计表提供 REST 查询。适合人群很明确：一类是做同主题毕设、想把模块边界和代码结构理清楚的同学；另一类是刚接手共享出行或物联网报表、准备自建存储链路的工程师。

2. 选型与存储分层：为什么“数据存储系统”不能只靠一个 MySQL

2.1 数据全链路：从骑行订单到统计报表，数据在哪个环节变成什么

共享单车数据存储系统的链路可以分成三个阶段，每个阶段解决一个独立问题：

第一个阶段是数据采集与落盘。单车智能锁上报的原始记录通常以 JSON 或 CSV 形式按天打包，一个文件少则几万条，多则几十万条。它们不会直接进 MySQL，而是先按日期分区存放在本地磁盘或 HDFS 上，作为“原始层”等待处理。

第二个阶段是清洗与聚合。由 Spark 批量作业读取当天原始文件，剔除坏数据，再按城市、小时、区域等维度做统计。这一步是系统里计算压力最大、代码最容易翻车的地方，也是引入 Spark 的最直接原因。

第三个阶段是接口服务。Spring Boot 连接 MySQL 中的统计表和清洗后的明细表，向运营后台或前端 H5 提供按日期查询骑行量、平均时长、营收等 REST 接口。这个阶段不需要陪 Spark 玩，只需要把 SQL 写对、把缓存命中率做上去。

下面这张表把三个阶段对应的数据处理方式和数据流向整理清楚：

数据从原始层流向聚合层时，会经过一次“不可逆的瘦身”。原始记录可能一个月有上亿条，但统计表按天加城市只有几千行，这部分容量差就是 Spark 在清洗和聚合过程中压缩掉的。所以“数据存储系统”并不是简单地把数据塞进一个数据库，而是要在不同层级选择不同的存储形态。

2.2 存储分层：原始层、聚合层、应用层分别放哪、怎么选引擎

原始层我一般不建议直接进 MySQL。骑行的原始日志文件以 JSON/CSV 落盘后，保留价值在于事后抽查和回溯清洗逻辑，查询频率极低。直接导入 MySQL 不仅浪费空间，而且会让 MySQL 的表数量膨胀到难以维护。常见做法是把原始文件按日期压缩成 gzip 格式归档，保留 30 天左右。

聚合层是这套系统的核心存储，使用 MySQL。共享单车的统计场景有几类典型查询：按天查总骑行量、按城市查各时段的骑行量、按运营区域查高峰期的车辆调度缺口。这些查询的特点是维度和条件固定，把结果预先聚合好写入表里，接口查询只需要做简单的过滤和排序。

应用层对性能要求最高，但数据量最小。这里会用 Redis 缓存前一天或前一周的聚合结果，避免高峰期运营后台反复刷新时把 MySQL 打崩。Redis 只作为二级缓存存在，缓存 key 设计为“日期 + 城市”，失效时间设为 2 小时左右即可。

选型时还有一个容易忽略的点：MySQL 表的引擎必须统一使用 InnoDB。有些同学为了省空间把历史表改成 MyISAM，结果跑着聚合查询时全表锁住，接口超时，这是一个很典型的毕设翻车现场。

2.3 技术选型理由：Spark 批处理、Spring Boot 接口、MySQL 落库，边界在哪

把 Spark 和 Spring Boot 放进同一个项目，很多人会问：数据量到底多大才需要 Spark？我的判断标准是：单日原始记录超过 20 万条、单文件超过 100 MB，Spring Boot 逐行解析就开始变得吃力。此时用 Spark 分批读取和并行聚合，代码复杂度并没有增加多少，但处理时间可以从半小时压缩到几分钟。

有人会想到用 Flink 做流式处理。共享出行的数据报表目前基本还是按天或按小时跑批，并不是真正需要毫秒级延迟的场景。Flink 的引入成本在于状态管理和精确一次语义的配置，在数据量没到实时报警级别前，用它属于杀鸡用牛刀。

Spring Boot 的边界非常清楚：只做查询接口，不做数据清洗。即使项目里有现成的 SparkSession，也不要在 Controller 里直接查 DataFrame。Spark 作业作为独立任务运行，结果落库后，Spring Boot 再对 MySQL 提供查询，这是最稳妥的工程化分工。

3. 用 Spark 做清洗与聚合落库：从 CSV/JSON 到 MySQL 的完整代码流程

3.1 Spark 环境初始化：Session 参数与本地/集群提交方式

我习惯在 Maven 工程里单独建一个 etl 模块，把 Spark 作业代码和 Spring Boot 代码分开编译。这样做的原因是：Spark 的依赖体积大，如果和 Web 模块的依赖混在一起，Spring Boot 启动时会因为 jar 冲突加载缓慢，甚至直接起不来。

先看最常用的 SparkSession 初始化写法：

import org.apache.spark.sql.SparkSession; public SparkSession buildSparkSession() { return SparkSession.builder() .appName("BikeTripDailyEtl") .master("spark://node01:7077") .config("spark.executor.memory", "2g") .config("spark.executor.cores", "2") .config("spark.sql.shuffle.partitions", "24") .config("spark.sql.session.timeZone", "Asia/Shanghai") .config("spark.network.timeout", "300s") .getOrCreate(); }

这段代码里最关键的是spark.sql.shuffle.partitions。如果不设置，Spark 默认按 200 个分区做 shuffle，对单日百万级的数据量来说很容易产生大量小块文件，拖慢后续写入 MySQL 的速度。我一般把分区数设在 24 到 48 之间，具体值取决于集群可用核心数。

master参数在本地调试时改成local[*]即可，但要记住提交到集群时不要写死local，否则每个 executor 都会在本机起 worker，集群资源完全用不上。提交作业时使用 spark-submit 命令：

spark-submit \ --class com.example.etl.BikeTripDailyEtl \ --master spark://node01:7077 \ --executor-memory 2g \ --total-executor-cores 8 \ bike-etl-1.0.jar \ --input /data/bike/2024-03-01

参数说明：--class指定主类，--executor-memory控制每个 executor 的堆内存，--total-executor-cores限制作业最大占用核心数。这两个参数一定不要超过集群实际资源，否则作业会一直卡在 Waiting 状态。--input是作业自定义参数，用于接收当天数据的目录路径。

3.2 读取骑行订单数据：spark.read 处理 JSON 和 CSV 的差异

共享单车数据源最常见的是 JSON 和 CSV 两种格式。很多同学在这里踩的第一个坑是：CSV 读出来经纬度全是字符串，做聚合时报类型错误。先看两段读取代码：

import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; Dataset<Row> rawJson = spark.read() .format("json") .option("multiline", "false") .option("encoding", "UTF-8") .load("/data/bike/2024-03-01/*.json"); Dataset<Row> rawCsv = spark.read() .format("csv") .option("header", "true") .option("sep", ",") .option("encoding", "UTF-8") .option("inferSchema", "true") .load("/data/bike/2024-03-01/*.csv");

JSON 读取时要注意multiline选项：如果每行是一个完整 JSON 对象，保持false；如果整个文件是一个大 JSON 数组，必须设为true，否则只能解析出第一行。CSV 的inferSchema会把数值列自动识别为 Long 或 Double 类型，但前提是文件里这个列没有混入空字符串。如果有空值，整列会被推断成 String，后续聚合时要做 cast。

我处理这类问题的标准姿势是先加载原始数据，打印 schema 确认类型：

rawCsv.printSchema(); rawCsv.show(10, false);

show(false)表示不截断长字段，能看清经纬度字段里是否混入的异常数据。这一步花几分钟，能省掉后面排查类型的很多时间。

3.3 清洗规则与聚合维度：把乱数据变成统计口径

共享单车原始数据的“脏”，不是脏在格式，而是脏在业务逻辑。比如：用户扫码后马上取消，产生一条时长 1 秒的订单；单车被异常锁住，订单时长可能超过 10 小时；GPS 信号漂移导致经纬度超出城市范围。这些脏数据会直接把平均时长和区域骑行的统计结果拉偏。

我常用的清洗规则如下：

import static org.apache.spark.sql.functions.*; Dataset<Row> cleaned = raw .filter(col("trip_id").isNotNull()) .filter(col("bike_id").isNotNull()) .filter(col("duration_sec").gt(10) .and(col("duration_sec").lt(4 * 3600))) .filter(col("start_lng").gt(70.0) .and(col("start_lng").lt(140.0)) .and(col("start_lat").gt(0.0) .and(col("start_lat").lt(60.0))) .withColumn("data_date", to_date(col("start_time"))) .withColumn("data_hour", hour(col("start_time")));

清洗逻辑说明：前三个 filter 分别是去掉空订单号、去掉单车编号为空的数据、过滤掉时长小于 10 秒或大于 4 小时的数据。最后两个 withColumn 把时间字段拆成“日期”和“小时”两列，方便后面按小时维度聚合。

这里要注意filter里多个条件的写法，Spark 的col("a").gt(10).and(col("a").lt(20))是链式调用，不能直接用&&。很多刚上手 Spark 的同学在这里报编译错误，就是因为把 Java 语法带进来了。

聚合维度按需求设计。最常用的按天 + 城市聚合：

Dataset<Row> dailyStats = cleaned .groupBy("data_date", "city_id") .agg( count("trip_id").as("total_trips"), round(avg("duration_sec"), 2).as("avg_duration"), round(sum("duration_sec") / 3600.0, 2).as("total_hours"), round(sum("amount"), 2).as("total_amount") );

3.4 写入 MySQL：Spark JDBC 写入的连接配置与 SaveMode 选择

Spark 写 MySQL 的底层机制是每个分区各建一个 JDBC 连接，写入的并行度与分区数直接相关。先看写入代码：

dailyStats.write().mode(SaveMode.Append) .format("jdbc") .option("driver", "com.mysql.cj.jdbc.Driver") .option("url", "jdbc:mysql://127.0.0.1:3306/bike_share?useSSL=false&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true") .option("user", "bike_admin") .option("password", "从配置中心读取，不要硬编码") .option("dbtable", "trip_daily_stats") .save();

SaveMode.Append是写统计表最安全的模式，每次只追加当天数据。这里有一个高频坑：如果前一天已经跑过同一批数据，第二天重新执行时 Append 会产生重复统计，导致报表翻倍。解决方式是在写入前先删除当天分区数据：

spark.read() .option("dbtable", "trip_daily_stats") .option("user", "bike_admin") .option("password", "your_password") .format("jdbc") .load() .write().mode(SaveMode.Overwrite) ...

更简单的做法是在 MySQL 表上把data_date和city_id建联合唯一索引，然后在 Spark 写入时用SaveMode.Append，如果重复，会直接报主键冲突，提前失败。虽然粗暴，但能保证数据不重复。

JDBC 写入还有一个调优点：在 url 里加rewriteBatchedStatements=true，这会让 JDBC 驱动把多条 insert 合并成一条批量语句提交。不加这个参数时，Spark 每写一条数据就发一次网络请求，写几万行会非常慢。

4. Spring Boot 对接存储层：数据访问层、REST 接口与缓存怎么配

4.1 项目结构与数据访问层：MyBatis-Plus 还是 Spring Data JPA

Spring Boot 模块和 Spark 模块在同一个 Maven 工程里时，项目结构我会按 controller、service、mapper、entity 四层组织。实体类里用 Lombok 的@Data注解减少 getter/setter，Mapper 接口继承 MyBatis-Plus 的BaseMapper，CRUD 基本不用自己写 SQL。

选 MyBatis-Plus 而不是 JPA 的原因是：统计查询场景中 SQL 条件变化频繁，MyBatis-Plus 的 LambdaQueryWrapper 写起来直白，团队里新人也容易上手。JPA 在处理复杂聚合查询时，需要写原生 SQL 或 JPQL，不如 MyBatis 直观。

依赖层面，pom 中需要加入的依赖大致如下：

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-boot-starter</artifactId> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <scope>runtime</scope> </dependency>

这里有一个版本敏感点：mysql-connector-java 的runtimescope 意味着编译时不依赖，但运行时必须有。如果 Spring Boot 版本较高，MySQL 连接串里用com.mysql.cj.jdbc.Driver，旧版本用的com.mysql.jdbc.Driver在新驱动里已经移除，这是常见的启动报错。

4.2 查询接口实现：按时间范围、区域、单车的三个端点

运营后台最常用的三个查询是：按时间范围查骑行量趋势、按区域查某个时段的骑行分布、按单车编号查它的历史骑行记录。先看一个按时间范围的 Controller：

@RestController @RequestMapping("/api/v1/trip-stats") public class TripStatsController { @Autowired private TripDailyStatsService statsService; @GetMapping("/daily") public List<TripDailyStats> daily( @RequestParam String startDate, @RequestParam String endDate) { return statsService.list( Wrappers.<TripDailyStats>lambdaQuery() .between(TripDailyStats::getDataDate, startDate, endDate) .orderByAsc(TripDailyStats::getDataDate) ); } }

这段代码没有写一行 SQL，因为 MyBatis-Plus 的 lambdaQuery 已经把条件封装好了。between对应 SQL 的 BETWEEN AND，orderByAsc按日期升序排列。注意TripDailyStats::getDataDate这种写法，编译期就能校验字段名，比字符串形式的"data_date"安全得多。

按车辆查历史记录的接口稍微复杂一点，要在 Service 层加一层判断：

public List<TripDetail> listByBike(String bikeId, String startDate, String endDate) { QueryWrapper<TripDetail> wrapper = new QueryWrapper<>(); wrapper.eq("bike_id", bikeId) .between("data_date", startDate, endDate) .orderByDesc("start_time"); return tripDetailMapper.selectList(wrapper); }

注意这里查询的是清洗后的明细表trip_detail，不是聚合表。明细表数据量大，必须给bike_id和data_date建联合索引，否则全表扫描会让接口直接超时。

4.3 缓存与连接池：Redis 和 HikariCP 的常见配置

接口层还有一个不可回避的问题：运营后台一个页面会同时发多个接口请求，如果每次都打 MySQL，高峰期连接池会被占满。我会把热点查询结果缓存到 Redis，缓存逻辑放在 Service 层而不是 Controller 层：

public List<TripDailyStats> listBetween(String startDate, String endDate) { String cacheKey = "trip_stats:daily:" + startDate + "_" + endDate; Object cached = redisTemplate.opsForValue().get(cacheKey); if (cached != null) { return (List<TripDailyStats>) cached; } List<TripDailyStats> result = statsService.list(...); redisTemplate.opsForValue().set(cacheKey, result, 2, TimeUnit.HOURS); return result; }

缓存穿透的防御点在 cacheKey 里，按日期范围组合自然分散了 key，不会出现固定热点 key 被击穿的情况。如果查询范围跨天超过 7 天，我一般直接绕过 Redis 查 MySQL，因为这种数据量不大但每次请求都会重算缓存，命中率太低。

HikariCP 连接池配置在application.yml中：

spring: datasource: hikari: maximum-pool-size: 20 minimum-idle: 5 connection-timeout: 3000 max-lifetime: 1800000

maximum-pool-size设为 20 是经过权衡的：Spark 写入时占用的连接数和 Spring Boot 查询时的连接数共用同一个池，如果池太大，Spark 批量写数据时可能把池占满，导致接口查询无连接可用。connection-timeout3 秒比较合理，超过即失败，能让异常请求快速暴露。

5. 避坑与排坑：Spark + Spring Boot 协作中的 5 个高频问题

5.1 现象：Spark 作业提交后 Container 反复退出，日志刷 OutOfMemoryError

原因：spark.executor.memory设成 2g，但集群每台 worker 的实际可用内存只有 1.5g，executor 分配不到足够的堆内存，容器一启动就被杀掉。也有人误把 driver 内存当成 executor 内存调，本地调试时没发现问题，提交集群就翻车。

解决：先看 worker 节点的空闲内存，free -h确认可用内存后，再把 executor 内存调到可用内存的 80% 以下。同时检查spark.executor.cores和spark.executor.memory的比例关系，每个 executor 核数不要超过 4，否则内存分配容易超过单机上限。

5.2 现象：Spark 写 MySQL 时抛出引发 Connection 拒绝的异常

原因：两个独立问题混在一起。第一，MySQL 的max_connections默认为 151，Spark 的 24 个分区同时写入时，每个分区建一个连接，加上 Spring Boot 应用占用的连接数，很容易打满。第二，Spark 作业所在节点和 MySQL 不在同一网段，防火墙拦截了 3306 端口。

解决：先把 MySQL 的max_connections调大到 300，再在 Spark 写入前执行.option("numPartitions", "8")，限制写入任务的并行连接数不超过 8。防火墙问题用telnet mysql主机ip 3306测试连通性，不通就找运维开白名单。

5.3 现象：聚合结果表里的经纬度数值比实际城市坐标整体小几度

原因：读入的文件里有部分是字符串类型的经纬度字段，Spark 在聚合时把它 cast 成 Double 进行计算，但字符串里的坐标格式是“度分秒”形式，不是十进制小数。部分坐标直接按字符串转数字，解析出的数值自然就偏了。

解决：在清洗阶段显式处理经纬度字段，统一转换成十进制小数。如果数据源里已经混入两种格式，最稳妥的做法是写一个 UDF 做格式转换，而不是依赖 Spark 的隐式类型转换。转换后加一轮边界检查，抽掉超出城市经纬度范围的记录。

5.4 现象：Spring Boot 版本太高导致项目启动时直接报错，找不到 Spark 相关类

原因：Spring Boot 3.x 从 Java 17 开始强制要求 javax 换成 jakarta 命名空间，而常见的 Spark 2.4.x 和 Scala 2.11 编译的 jar 仍然基于 javax。把 Spring Boot 升到 3.x 后，Spark 引用的javax.servlet类在容器里找不到，应用启动失败。

解决：做这套系统最稳妥的组合是 Spring Boot 2.7.x 搭配 Spark 3.1.x，JDK 用 8 或 11。版本选择上不要追求最新，稳定兼容才是第一位的。如果确实要用高版本 Spring Boot，就需要把数据处理作业拆成独立进程，用 spark-submit 提交，这样 Web 容器和 Spark 环境完全分离，互相不干扰。

5.5 现象：MySQL 表里中文数据全部变成问号

原因：建表语句没有指定utf8mb4字符集，Spark 写入的中文被按默认的 latin1 编码存储，读出来全部变成乱码。这个问题在 Spring Boot 查询时不容易发现，因为它读到的“乱码”实际上已经是库里的固定值。

解决：建表时统一指定字符集，例如CREATE TABLE ... DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci。对于已经建好的表，执行ALTER TABLE trip_daily_stats CONVERT TO CHARACTER SET utf8mb4可以修复。同时 Spark JDBC url 里要加characterEncoding=UTF-8，保证写入时按 UTF-8 编码传输。

6. 定时调度与增量写入：让 ETL 从手动跑批升级为每日任务

大部分毕设做到 Spark 清洗落库、Spring Boot 查询就已经收工了，但真正能用的系统还需要解决一个问题：数据每天产生，不可能每次都手动跑一次 spark-submit。

我的做法是用 Spring Boot 自带@Scheduled注解做定时调度，每天凌晨 1 点 30 分触发当天的 ETL 作业：

@Component public class EtlScheduler { @Autowired private SparkEtlRunner sparkEtlRunner; @Scheduled(cron = "0 30 1 * * ?") public void runDailyEtl() { LocalDate yesterday = LocalDate.now().minusDays(1); String path = "/data/bike/" + yesterday; sparkEtlRunner.run(path, yesterday.toString()); } }

cron 表达式0 30 1 * * ?表示每天 1 点 30 分执行，选这个时间是为了避开共享单车凌晨的骑行低谷和数据落盘延迟。SparkEtlRunner是封装好的 Spark 作业入口，接收目录路径和日期两个参数。

增量写入的逻辑是核心：每次只读取/data/bike/2024-03-01/这种带日期的目录，Spark 读的就是当天的文件，处理后也只写入当天分区的数据，天然实现了增量。不需要在代码里维护偏移量，也不需要引入额外的消息队列。

验证增量是否跑通的习惯动作是查两张表：先查源目录文件总数，再查目标表里当天的记录数，两者应当匹配。更严谨一点，在 ETL 结束时把当天的 total_trips 写入一张 job_history 表，第二天通过对比 job_history 里连续两天的 total_trips 是否有跳变，能快速发现数据缺失。

我个人的教训是：数据管道里最不靠谱的就是靠人脑记录进度，一定要让作业本身可查询、可重跑。每天跑批前先检查前一天作业是否成功，失败时直接手动重跑当天日期目录，不用担心造成重复，因为写入模式是 Append 加唯一索引。这套流程跑顺之后，系统才真正从一个毕设项目变成了能日常使用的数据存储与分析工具，希望这些思路能帮到你。

本文还有配套的精品资源，点击获取