简介：一套基于Spark的地铁大数据客流分析系统毕业设计源码包，面向计算机相关专业学生、大数据初学者及毕业设计选题者，围绕地铁客流统计、预测与调度优化场景展开。压缩包共194个文件，约42.6MB，按功能分为Java/Scala核心源码、Spark作业配置、SQL数据库脚本、CSV样例数据、系统设计文档及可视化素材等，目录结构清晰，便于按模块检索。目前已有259人学习下载。资源完整覆盖需求分析、系统设计、编码实现与测试部署流程，涉及Spark SQL结构化查询、Spark Streaming实时处理、MLlib机器学习预测等关键技术，并整合HBase存储、Logstash日志接入及数据预处理环节，可帮助读者快速理解分布式大数据分析项目的工程化方法。附带的源码注释与测试数据更适合动手实践，是完成毕业设计或入门Spark开发的实用参考资料。

1. 基于Spark的地铁大数据客流分析：从CSV原始数据到分时报表完整链路

把几千万条地铁刷卡记录交给你，你能在多长时间内算出“早高峰哪个站进站量最大”这种问题？某地铁运营部门之前拿 Excel 处理这类需求，一次要等一两个小时，文件超过两百万行就直接卡死。这份基于 Spark 的地铁大数据客流分析系统，就是专门为这种场景写的毕设级完整项目：从原始 CSV 客流数据出发，经过清洗、聚合，落到 HBase，再通过查询接口输出分时客流、断面客流和预测结果。适合正在做大数据方向毕业设计的学生，也适合想快速搭一套 Spark + HBase 离线分析 demo 的开发者。压缩包里源码、配置、样例数据、接口调试脚本都齐，照着跑就能看完整链路。

2. 架构与技术选型：四层结构里为什么偏偏是 Spark 配 HBase

2.1 系统分层：采集、存储、计算、服务各干各的

这套系统不是单机脚本，而是按生产级大数据项目的套路拆成了四层。第一层是采集层，负责把数据弄进来：一批是 CSV 批量导入，比如压缩包里的 szmc.net-metro.csv 这类历史客流数据；另一批是通过 Logstash 采集 nginx 网关的访问日志，用来跟踪数据更新和接口调用情况。第二层是存储层，原始文件落在 HDFS 上，聚合后的结果写进 HBase，这样既能跑批又能支撑点查。第三层是计算层，Spark SQL 做离线统计、Spark Streaming 做准实时客流、MLlib 做客流预测，三者共用同一套 HBase 表。第四层是服务层，对外提供 HTTP 查询接口，search.http 和 szt-api.http 就是调试这套接口用的请求脚本，前端或报表工具拿接口数据渲染图表。

四层之间通过 HDFS 路径和 HBase 表解耦：采集任务只负责往 HDFS 放文件，计算任务监听路径或定时触发，服务层只查 HBase。哪怕某天把 CSV 换成 Kafka 数据源，也只需要改采集层和计算层入口，HBase 表结构不用动。

2.2 技术选型：MySQL 加 Python 为什么扛不住这个场景

如果只是几十万行数据，MySQL 加 Python 完全够用，写个 group by 就出来了。但地铁客流数据的特点是量大、字段杂、查询要求快：一天上千万条交易记录，按站点、时段、线路多个维度反复聚合，还要支持“查某站某天 8 点到 9 点进站量”这类高频点查。

结论很直接：Spark 负责“算得动”，HBase 负责“查得快”。Spark 把几千万条数据分布式聚合成结果集，再按设计好的 RowKey 写进 HBase，查询接口拿到站点和时间参数后走 RowKey 精确匹配，整个过程不扫全表。

2.3 核心数据模型：HBase 表设计与 RowKey 规划

项目里最关键的表是地铁客流聚合表，主要存“某线路某站点某时段进出站量”这种统计数据。HBase 表设计如下：

RowKey 设计是这套系统的关键。如果直接拿线路编码开头，数据会全部压在同一台 RegionServer 上，形成热点。常见做法是在 RowKey 前拼两位哈希盐值，再拼线路、站点、日期、时段，让写入分散到多个 Region，查询时按盐值加已知条件组合成前缀扫描。建表时最好预分区，避免 Region 频繁分裂，HBase shell 下的建表命令大致长这样：

create 't_metro_flow', {NAME => 'cf', COMPRESSION => 'SNAPPY', BLOOMFILTER => 'ROW'}, {NUMREGIONS => 16, SPLITALGO => 'UniformSplit'}

这段命令里，COMPRESSION 设成 SNAPPY 是为了压缩磁盘占用，LZO 也能用但 SNAPPY 在 Spark 和 HBase 两边都内置，少一层依赖。BLOOMFILTER 设成 ROW 类型，点查时能跳过大量不包含目标 RowKey 的 HFile。NUMREGIONS 按预估数据量调，一般按数据量除以每个 Region 的合理容量来定，16 个 Region 对毕设量级已经够用。SPLITALGO 用 UniformSplit 表示按 RowKey 范围均匀切分。

3. 目录结构与数据流转：解开 Graduation Design 压缩包里的每一份配置

3.1 压缩包里的文件逐个看

拿到压缩包，打开后不只是源码，还有配置文件、样例数据和调试脚本。把这些文件按作用理一遍，整个系统的拼图就清楚了：

注意 szmc.net-metro.csv 这一类文件，名字里带城市前缀，内容本身是一份脱敏后的客流样例，不是让你原样当成生产数据用。做真实项目时，把文件名的前缀和表结构里的线路字典换成自己城市的即可。

3.2 logstash-nginx.config：日志采集入口怎么配

Logstash 在这个项目里承担的是辅助采集角色，重点是 nginx 访问日志。配置一般分 input、filter、output 三段，项目里的 logstash-nginx.config 核心逻辑大致如下：

input { file { path => "/var/log/nginx/metro-access.log" start_position => "beginning" sincedb_path => "/dev/null" } } filter { grok { match => { "message" => "%{IPORHOST:client_ip} %{GREEDYDATA:request}" } } date { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] } } output { stdout { codec => json_lines } }

这段配置里，start_position 设成 beginning 表示从文件头开始读，适合第一次接入历史日志；sincedb_path 指向 /dev/null 是为了每次重启 Logstash 都重新读一遍，避免因为 sincedb 记录偏移导致漏数据，毕设调试阶段特别好用。grok 里的 IPORHOST 和 GREEDYDATA 是正则模式，分别匹配客户端 IP 和请求路径。date 插件把日志里的时间字符串解析成标准时间戳，后续按小时窗口聚合就靠这个字段。output 先打成 json_lines 打到控制台，确认日志格式没问题后再改成输出到 Elasticsearch 或直接落 HDFS。

3.3 样例数据字段与清洗思路

szmc.net-metro.csv 这类地铁客流数据，字段一般包含交易卡号、线路编码、站点编码、进出站标志、交易时间等。原始数据直接进 Spark 之前必须清洗，压缩包里虽然没有单独的清洗脚本，但最合理的做法是用 pandas 做一次预处理，减少 Spark 作业里处理脏数据的负担：

import pandas as pd df = pd.read_csv("szmc.net-metro.csv", dtype={"卡号": str}) df["交易时间"] = pd.to_datetime(df["交易时间"]) df = df.dropna(subset=["站点编号", "进出站标志"]) df["时段"] = df["交易时间"].dt.hour.apply(lambda h: f"{h:02d}:00-{h+1:02d}:00") df.to_csv("metro_clean.csv", index=False, encoding="utf-8")

这里 dtype 把卡号强制转成字符串，是为了防止长数字被 pandas 自动读成科学计数法丢掉精度，这是刷卡数据最常见的坑。dropna 只保留站点编号和进出站标志两个关键字段非空的行，脏数据大多集中在这两列。时段字段按交易时间的小时切分，后续 Spark SQL 聚合直接按这个字段分组就行。清洗完的文件再上传 HDFS，避免 Spark 每次作业都重复处理异常值。

4. 核心模块拆解：离线聚合、实时统计、预测与 HBase 写入的完整代码

4.1 Spark SQL 离线聚合：统计分时进出站量

离线统计是整套系统的地基。把清洗后的 CSV 读成 DataFrame，注册临时表，用 SQL 按线路、站点、时段分组聚合，这是 Spark SQL 最典型的用法，也是这套系统里最核心的一段逻辑：

val spark = SparkSession.builder() .appName("MetroFlowOffline") .enableHiveSupport() .getOrCreate() val df = spark.read .option("header", "true") .csv("hdfs:///data/metro/metro_clean.csv") df.createOrReplaceTempView("raw_flow") val result = spark.sql(""" SELECT line_code, station_code, period, SUM(1) AS flow FROM ( SELECT line_code, station_code, concat( date_format(trade_time, 'HH:00'), '-', date_format(trade_time, 'HH:59') ) AS period FROM raw_flow WHERE in_out_flag IN ('IN', 'OUT') ) t GROUP BY line_code, station_code, period """) result.write.mode("overwrite").saveAsTable("ods_metro_flow")

这段代码里，enableHiveSupport 是为了让 Spark 能读写 Hive 元数据表，毕设如果没配 Hive 可以去掉这行，改成直接 write.parquet 落到 HDFS。date_format 把交易时间格式化成小时字符串，再拼成“08:00-08:59”这种时段标签，保证 Group By 的分组键是可读的。WHERE 条件里 in_out_flag 过滤出进出站记录，去掉异常类型。最后 saveAsTable 会把聚合结果注册成表，方便预测模块直接读。

4.2 Spark Streaming 准实时统计：窗口内客流走势

离线结果只能看昨天或上一小时，想盯当前十分钟的进站高峰，需要流处理。项目里常见的实现是 Spark Streaming 从 socket 或消息队列收实时刷卡记录，按固定窗口聚合：

val ssc = new StreamingContext(sparkConf, Seconds(10)) val lines = ssc.socketTextStream("localhost", 9999) val flows = lines.map(_.split(",")) .filter(arr => arr.length >= 3) .map(arr => ((arr(0), arr(1), computePeriod(arr(2))), 1L)) val windowed = flows.reduceByKeyAndWindow( _ + _, Seconds(300), Seconds(60) ) windowed.foreachRDD { rdd => rdd.foreachPartition { part => HBaseWriter.write(part) } } ssc.start() ssc.awaitTermination()

这段逻辑里，socketTextStream 从 9999 端口收数据，方便本地测试，生产环境可以换成 Kafka。reduceByKeyAndWindow 的三个参数分别是聚合函数、窗口长度和滑动间隔，这里窗口 300 秒、滑动 60 秒，表示每 60 秒输出最近 5 分钟的聚合结果。foreachPartition 里每个分区单独建 HBase 连接批量写入，避免每条记录都创建一次连接，这是流写入 HBase 的性能关键。

4.3 HBase 批量写入与参数调优

离线聚合结果写 HBase，最怕逐条 put 导致写入慢、RegionServer 压力大。日常处理这种批量写，通常用 BufferedMutator 攒一批再刷，代码大致如下：

val putList = new java.util.ArrayList[Put]() result.collect().foreach { row => val rowKey = s"${salt(row.getAs[String]("station_code"))}_" + s"${row.getAs[String]("line_code")}_" + s"${row.getAs[String]("period")}" val put = new Put(Bytes.toBytes(rowKey)) put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("flow"), Bytes.toBytes(row.getAs[Long]("flow").toString)) putList.add(put) if (putList.size() >= 500) { hTable.put(putList) putList.clear() } } if (!putList.isEmpty) hTable.put(putList)

按 500 条一批提交，每批刷一次，比逐条 put 快一个数量级。RowKey 里第一个字段我加了盐值，把不同站点的数据分散到不同 Region。如果数据量再大一个级别，可以选择 HBase 的 BulkLoad 方式，先按 HFile 格式生成文件再直接加载到表里，跳过写 WAL 和 memstore 的过程，适合一次性灌历史数据。毕设阶段用 BufferedMutator 已经足够。

4.4 客流预测：用 MLlib 线性回归做基线模型

预测模块最常用的做法是拿历史分时客流训练模型，预测未来同时间段客流。项目里选择线性回归做基线，简单可解释，跑通后再换更复杂的模型:

import org.apache.spark.ml.regression.LinearRegression import org.apache.spark.ml.feature.VectorAssembler val train = spark.sql(""" SELECT hour_index, flow FROM ods_metro_flow WHERE station_code = 'A001' AND line_code = 'L1' AND dt < '2024-03-01' """) val assembler = new VectorAssembler() .setInputCols(Array("hour_index")) .setOutputCol("features") val lr = new LinearRegression() .setMaxIter(100) .setRegParam(0.01) val model = lr.fit(assembler.transform(train)) model.write.overwrite().save("/model/metro/L1_A001")

这里 hour_index 是小时编号，比如 8 点对应 8，flow 是历史同时段客流，特征只有一维，精度有限但胜在跑得快。setRegParam 设 0.01 是给模型加一点 L2 正则，防止在少量训练数据上过拟合，这是毕设里容易忽略的参数。模型保存到指定路径后，预测任务每天定时读取前一天数据，生成当天分时预测写回 HBase 的预测表，接口层把预测值和实际值一起返回，前端就能画对比曲线。

5. 常见问题排查与避坑记录：从环境崩溃到数据倾斜的五次翻车现场

5.1 环境准备与启动顺序

跑这套系统，本地环境建议是 JDK 8、Scala 2.12、Spark 2.4 或 3.x 均可，HBase 用 1.4 或 2.x，Hadoop 选与 Spark 兼容的版本组合。启动顺序有讲究：先启动 Hadoop 的 HDFS，再启动 HBase，最后才提交 Spark 作业。HDFS 没起来就去跑 Spark 写表，大概率会卡在连接元数据上。

# 1. 起 HDFS start-dfs.sh # 2. 起 HBase start-hbase.sh # 3. 建表 hbase shell hbase.command # 4. 跑离线任务 spark-submit \ --class com.metro.offline.MetroFlowOffline \ --master yarn \ metro-offline.jar # 5. 起查询服务 java -jar metro-api.jar --server.port=8080

提交 Spark 任务时，--master 如果用 local 模式跑通了，不代表集群模式没问题。最省事的方式是先在 local[2] 下把任务调通，再切 yarn，这样可以把业务逻辑问题和环境问题分开排查。

5.2 怎么确认任务真的跑通了

Spark 任务日志一大堆，新手经常看到 INFO 刷屏就以为成功了。判断标准是看三处：一是 YARN 页面上 Application 状态变成 SUCCEEDED；二是目标表 t_metro_flow 里能查到聚合结果；三是接口服务日志里出现 200 响应。光看控制台 Stage 进度条走完是不够的，任务可能在整个 Job 的最后一步写数据时失败。

5.3 五条血泪踩坑记录

第一坑：提交作业直接 NoSuchMethodError，Spark 和 Scala 版本对不上。现象是作业一启动就抛 java.lang.NoSuchMethodError，提示某个 scala 包里的方法不存在。原因是本地编译用的 Scala 2.12，集群上 Spark 自带的是 Scala 2.11，两边二进制不兼容。解决方法是统一编译环境和运行环境的 Scala 版本，或者在 pom.xml 里把 scala-library 依赖的 scope 设为 provided，交给 Spark 运行时提供。

第二坑：HBase 写入热点，一台 RegionServer 负载飙高。现象是写入任务跑得慢，某个 RegionServer 的 CPU 长期 100%，其他节点闲着。原因是 RowKey 直接拿线路编码开头，数据全挤在少数 Region。解决方法是 RowKey 前拼两位盐值，比如站点编码 hashCode 取模后补零，让数据均匀散到不同 Region，代价是查询时要多带一个条件算出盐值前缀。

第三坑：CSV 读出来中文全乱码。现象是 Spark 读进 DataFrame 后，站点名称显示成乱码，聚合结果没法看。原因是样例数据是 GBK 编码，Spark 默认按 UTF-8 读。解决方法是读文件时加 option 指定编码，写成 spark.read.option("encoding", "GBK").csv(...)，或者清洗阶段统一转成 UTF-8 再上传。

第四坑：离线任务 OOM，Executor 直接挂掉。现象是 Executor 日志频繁出现 GC 超时，Container 被杀。原因是聚合前数据量太大，某个站点数据特别多形成倾斜，加上默认内存只有 1G。解决方法是调大 spark.executor.memory，同时对热点 key 加随机前缀打散再做二次聚合，等于先本地聚合一轮再全局聚合。

第五坑：HBase shell 连不上，报 Connection refused。现象是执行 list 命令超时，客户端连不上集群。原因是本地调试时只启动了 HBase 单进程，或者 ZooKeeper 地址配成了 localhost 而服务端监听在内网 IP。解决方法是先跑 hbase zkcli 确认 ZooKeeper 正常，再核对客户端 hbase-site.xml 里的 quorum 地址是否和集群实际监听地址一致。

注意：这些坑没有一个是“玄学”，全部能从日志和配置里找到根源，排查时按日志时间戳倒序看错误栈，比乱试参数有效得多。

6. 二次开发思路：把单城客流系统改造成可扩展的分析底座

6.1 从 CSV 换到 Kafka：离线分析升级成实时链路

现在这套系统是 CSV 导入加 Spark Streaming socket 测试，改成生产可用只需要替换数据入口。用 Kafka 作为采集层缓冲，Logstash 或 Flume 把刷卡记录送进 Kafka topic，Spark Streaming 通过 createDirectStream 消费指定 topic，断点续传和分区并行都交给 Kafka 处理。改动集中在采集层和流任务入口，HBase 表结构不需要变。

6.2 增加断面客流与满载率计算

断面客流是地铁运营真正关心的指标，计算思路是拿进出站记录做 OD 匹配，得到每个乘客的乘车区间，再统计每个区间段上同一时间经过的乘客总量。最简单的一版可以基于 Spark SQL 实现：同一张卡同一个时间段内的进站和出站记录关联成 OD 对，再把 OD 对和线路站点顺序表映射到断面，最后按断面聚合得到满载率。这段逻辑建议独立成单独模块，结果表单独设计，因为它的查询模式和分时客流完全不同。

6.3 让查询接口更通用：参数化设计与索引优化

查询接口目前是按站点和时段查，改造方向是支持多条件组合，比如线路加日期范围加时段区间。接口层把查询参数拼成 HBase Scan 的起始 RowKey 和结束 RowKey，走前缀扫描。如果出现按站点反向查线路的需求，就建一张站点到线路的映射表，或者加二级索引。预测结果和实际结果建议分表存储，接口层做一次 merge，避免混合写入导致 RowKey 设计互相妥协。

之前有一次我把 RowKey 设计成日期开头，结果某天的数据全打到一个 Region，线上查询从毫秒级退化到秒级，最后被迫做数据迁移。从那以后，我每次给 HBase 设计表结构，都强制先按数据分布画一遍读写路径，再写建表语句，这份客流系统的二次开发你也按这个顺序来，至少能避开我最疼的那次教训，希望帮到你。

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