Spark+HBase+Logstash地铁客流分析系统毕设源码全解析

简介：这是一套面向计算机类毕业设计的地铁大数据客流分析系统完整项目，以Apache Spark为核心，解决客流统计、趋势预测及调度优化问题。压缩包内按工程结构组织，涵盖需求分析文档、系统设计文档、Java/Scala源代码、Spark作业与集群配置、数据库建表与查询脚本，以及测试数据集和结果展示界面，覆盖数据采集、清洗、存储、分析到可视化全链路。资源共194个文件，以png界面截图、java与scala源码、xml配置、sql数据库脚本为主，附带csv样本数据、sh部署脚本和yaml配置等，整体42.6MB，适合大数据、数据工程方向学生作为毕业设计参考或课程项目实战素材。当前已有259人学习下载。通过完整源码与配套文档，可快速掌握Spark SQL、Streaming和MLlib的实际用法，并理解HBase、MySQL等存储方案在客流场景中的落地，便于二次开发或完成毕业设计答辩。项目目录清晰，代码注释与配置说明齐备，适合按模块逐步研读和复用。

1. 这套毕设源码解决的是地铁高峰期“哪个站最挤”的问题

地铁一天产生上百万条刷卡和进出站数据，真正有价值的问题却很简单：早高峰哪个站人最多、晚高峰哪条线最堵、明天某站客流会涨还是跌。这套基于 Spark 的地铁大数据客流分析系统，就是围绕这几个问题组织的一套完整毕业设计项目，从数据采集、清洗、存储到离线计算和 API 查询，全链路都有代码覆盖。它不是概念演示，而是一个能真正跑起来、能看到统计结果和预测曲线的工程。适合正在选毕设方向的在校生，也适合想了解 Spark + HBase 落地方案的后端开发者。源码包里除了 Spark 作业，还带了 HBase 建表命令、Logstash 采集配置、CSV 样本数据和 HTTP 接口调试文件，一次能看清整条数据链路的组织方式。

2. 技术选型与整体架构：Logstash 进、HBase 存、Spark 算、HTTP 出

2.1 从文件列表反推技术栈

拿到压缩包先别急着解压，扫一眼文件名就能知道这套系统的技术骨架。核心几个文件的信息密度非常高：

这里面最容易被忽略的是 .http 后缀文件。它是 IntelliJ IDEA HTTP Client 的请求描述文件，说明这套系统不是只跑一个 Spark 任务就结束，而是把计算结果通过接口暴露出来，做成一个可查询的服务。整套链路从数据采集到前端查询是闭环的，这也是毕业设计里最容易拿高分的一种组织形式，因为评委会看到你懂“全流程”而不是只会调 API。

2.2 数据管道分段：采集、存储、计算、服务

整套系统按数据处理顺序分四层，每层职责非常清楚：

采集层主要由 Logstash 负责，读取 szmc.net-metro.csv 这样的离线样本文件，做字段解析、类型转换，然后写入 HBase。如果接真实场景，可以把 input 换成 Kafka 或者直接监听闸机日志文件，配置结构不变。

存储层用的是 HBase。客流数据的特点是列多、稀疏、按时间维度持续追加，HBase 的宽表模型天然适合存这类数据。hbase.command 里定义了表结构、列族和预分区策略，具体设计下一章拆。

计算层是整套系统的主角，Spark 作业从 HBase 读取原始客流记录，完成站点聚合、分时段统计、线路热度排名，再用 MLlib 做简单的客流趋势预测。计算结果会写回 HBase 或者 MySQL，供接口层查询。

服务层通过 search.http 和 szt-api.http 对外暴露 HTTP 接口，比如“查某站某天分时段客流”“查全网站点热度排行”。接口层不直接扫全表，而是查 Spark 预计算好的结果，保证响应速度。

2.3 为什么偏偏是 Spark + HBase + Logstash 这套组合

选型逻辑其实也代表了大数学习题的标准答案。Flume 虽然也能做采集，但 Logstash 的 csv filter 插件处理结构化文本更快，配置也更直观，改字段映射不用写代码。HBase 胜在 rowkey 设计灵活，按“站ID_日期_时段”组织数据后，Spark 读取时只需要扫描连续区间，不需要全表扫描。而 Spark 负责批量计算，SQL 表达聚合逻辑比纯 MapReduce 少写太多代码；MLlib 里现成的线性回归、决策树模型可以直接用来做客流预测，不用自己推导数学公式。

这套组合的代价是版本兼容问题多，后面第 5 章我会专门讲怎么踩坑。但作为毕业设计，它展示的知识面足够广：NoSQL 数据库、大数据采集、分布式计算、机器学习、API 开发，五个点都覆盖到了，答辩时每个方向都有东西可讲。

3. 配置文件拆解：四个关键文件看懂整套系统

3.1 hbase.command：表结构与 rowkey 设计

hbase.command 内容大概率是 HBase Shell 的建表脚本。客流分析最核心的表是明细表，常见建表逻辑如下：

# 创建客流明细表，列族 cf 开启 SNAPPY 压缩 create 'metro:passenger_flow', {NAME => 'cf', COMPRESSION => 'SNAPPY', VERSIONS => 1}, {NUMREGIONS => 32, SPLITALGO => 'HexStringSplit'}

这段命令的逻辑值得细看。表名 metro:passenger_flow 用命名空间区分业务域；列族叫 cf，存储站点名称、进站人数、出站人数等字段。SNAPPY 压缩是为了减少 HDFS 上的存储占用，地铁客流数据量大，一天几十万条，压缩比通常能到 3:1 以上。

预分区是这条命令里最关键的部分。默认情况下 HBase 一张表只有 1 个 Region，写入压力全打在一个节点上。这里指定 32 个 Region，配合 HexStringSplit 按 rowkey 哈希散列，写入时可以分散到多台 RegionServer。我把 rowkey 设计成“站ID_日期_时段”格式，比如0101_2024-06-01_07，这样同一站点同一天的数据在物理存储上连续，Spark 按站点扫数据时是顺序读，速度比随机读快一个数量级。

3.2 logstash-nginx.config：CSV 是怎么进到 HBase 的

这个文件是 Logstash 的管道配置，负责把地铁客流 CSV 文件解析成结构化记录并写入 HBase。配置核心逻辑如下：

input { file { path => ["/data/metro/szmc.net-metro.csv"] start_position => "beginning" sincedb_path => "/dev/null" mode => "read" } } filter { csv { columns => ["station_id", "station_name", "date", "time_period", "inbound", "outbound"] separator => "," skip_header => true } mutate { convert => { "inbound" => "integer" "outbound" => "integer" } } } output { hbase { table => "metro:passenger_flow" cf => "cf" rowkey => "%{station_id}_%{date}_%{time_period}" columns => [ "cf:station_name", "%{station_name}", "cf:inbound", "%{inbound}", "cf:outbound", "%{outbound}" ] } }

几个参数说明一下。input 里的 sincedb_path 设为 /dev/null，表示每次启动都从头读文件，这个配置在反复测试数据导入时很省心，否则 Logstash 会从上次读到的地方继续，造成数据反复读不进去的错觉。filter 里的 csv 插件用 columns 声明列名，skip_header 跳过 CSV 表头，mutate 把进站出站人数从字符串转成整数，避免 Spark 那边再做一次类型转换。

output 里的 hbase 插件需要额外安装，不是 Logstash 自带的，这里埋了一个坑，后面避坑章节会细说。rowkey 直接引用 filter 里解析出来的字段拼接，和 hbase.command 里预分区设计完全对应。

3.3 search.http / szt-api.http：服务层接口长什么样

这两个文件描述的是系统对外提供的查询接口，用 IntelliJ HTTP Client 的格式编写，内容类似这样：

### 查询指定站点某天分时段客流 GET http://localhost:8080/api/v1/flow/station?stationId=0101&date=2024-06-01 Accept: application/json ### 查询全网站点某天客流热度排行 GET http://localhost:8080/api/v1/flow/hot?date=2024-06-01&limit=10 Accept: application/json

注意两个接口的定位差异。第一个接口是明细查询，返回某个站点一天 24 个时段分别有多少人进出，数据粒度细，适合前端画折线图；第二个接口是排行榜查询，返回全网最挤的 10 个站，适合画柱状图或者热力图。

这类接口背后查的其实是 Spark 预计算好的结果表，不是直接去扫 HBase 明细表。原因很现实，Spark SQL 跑一遍一个小时的数据聚合作业要二十多秒，用户的浏览器等不了，所以离线任务算完写结果表，在线接口只做同步查询，这是典型的分层查询架构。.

3.4 szmc.net-metro.csv：样本数据字段解读

CSV 是这套系统的灵魂输入，文件名的 szmc.net-metro 是某地铁网络的数据导出标识，具体对应哪个城市不重要，字段结构才是关键。样本数据大概长这样：

字段结构本身不难，但它决定了后续所有计算逻辑怎么写。Spark 聚合时按 station_id 和 time_period 分组，预测时把 date 拆成星期几作为特征，这些全都是围绕这个 CSV 的字段设计展开的。如果你拿到的是真实数据但没有中文站名，只需要改 CSV 里 station_name 列的值，不需要动任何代码。

4. 核心实现：Spark 作业从读 HBase 到出预测结果的完整链路

4.1 从 HBase 读数据：Spark 作业的入口

Spark 读取 HBase 有两种常见写法，一种是用newAPIHadoopRDD操作底层 API，另一种是讲 HBase 数据映射成 DataFrame 后用 Spark SQL 处理。毕设场景里推荐后者，因为后续聚合、预测都要用 SQL 和 MLlib，DataFrame 是统一入口。

import org.apache.spark.sql.SparkSession import org.apache.hadoop.hbase.{HBaseConfiguration, TableName} import org.apache.hadoop.hbase.mapreduce.TableInputFormat import org.apache.hadoop.hbase.util.Bytes import org.apache.hadoop.hbase.client.Result import org.apache.hadoop.hbase.io.ImmutableBytesWritable val spark = SparkSession.builder() .appName("MetroFlowAnalysis") .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") .getOrCreate() val hbaseConf = HBaseConfiguration.create() hbaseConf.set(TableInputFormat.INPUT_TABLE, "metro:passenger_flow") val hBaseRDD = spark.sparkContext.newAPIHadoopRDD( hbaseConf, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result] ) val flowDF = hBaseRDD.map { case (_, result) => val rowkey = Bytes.toString(result.getRow) val fields = rowkey.split("_") val stationId = fields(0) val date = fields(1) val period = fields(2).toInt val stationName = Bytes.toString(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("station_name"))) val inbound = Bytes.toLong(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("inbound"))) val outbound = Bytes.toLong(result.getValue(Bytes.toBytes("cf"), Bytes.toBytes("outbound"))) (stationId, stationName, date, period, inbound, outbound) }.toDF("station_id", "station_name", "date", "time_period", "inbound", "outbound") flowDF.createOrReplaceTempView("flow")

这段代码里值得讲两个点。第一是配置了 Kryo 序列化，Spark 默认的 Java 序列化在传输 RDD 时性能差，换成 Kryo 后序列化后的数据体积能小一半以上，处理 HBase 返回的 Result 对象时效果尤其明显。第二是 rowkey 解析用了 split("_")，这就要求第 3 章里 Logstash 写入时的 rowkey 格式必须严格一致，前后端设计是联动的。

newAPIHadoopRDD 扫描时读的是整个表，实际生产环境会用 Scan 加 startRow 和 stopRow 限制范围，比如只扫某一天的数据。毕设阶段数据量不大，全表扫描能跑通，但答辩时如果能说出这个优化点，印象分会高不少。

4.2 分时段客流聚合：核心业务逻辑

拿到了 DataFrame，接下来就是客流分析的主要需求：按站点、按小时段统计进出站人数，并找出全网峰值时段。这个逻辑用 Spark SQL 表达非常直接：

val hourlyFlow = spark.sql(""" SELECT station_id, station_name, time_period, SUM(inbound) AS total_inbound, SUM(outbound) AS total_outbound, SUM(inbound + outbound) AS total_flow FROM flow WHERE date = '2024-06-01' GROUP BY station_id, station_name, time_period ORDER BY total_flow DESC """) hourlyFlow.show(20, false) val peakHour = spark.sql(""" SELECT time_period, SUM(inbound + outbound) AS flow FROM flow WHERE date = '2024-06-01' GROUP BY time_period ORDER BY flow DESC LIMIT 1 """)

第一个查询算的是每个站在每个小时段的总客流，也就是“哪个站在哪个时段最挤”。第二个查询从全站维度算全网全天哪个时段客流总量最大，这里能看到最晚高峰落在几点，对排班和调度有直接参考价值。

如果你接手做二次开发，最常改的就是这段 SQL。比如想按线路维度统计，只要在表里加一个 line_id 字段，再把这句 SQL 的 GROUP BY 改成 line_id, time_period 就行。想分析周末和工作日差异，把 WHERE 条件加上 dayofweek 过滤即可。

4.3 用 MLlib 做客流趋势预测

预测模块是 Spark 引入 MLlib 的核心卖点，毕设里通常用线性回归做基线模型，预测某个站未来几个小时的客流。特征不能直接用日期字符串，得先转成数值：

import org.apache.spark.ml.feature.VectorAssembler import org.apache.spark.ml.regression.LinearRegression val historyDF = spark.sql(""" SELECT day_of_week, time_period, inbound FROM flow WHERE station_id = '0101' ORDER BY date, time_period """) val assembler = new VectorAssembler() .setInputCols(Array("day_of_week", "time_period")) .setOutputCol("features") val trainDF = assembler.transform(historyDF) .select("features", "inbound") .withColumnRenamed("inbound", "label") val lr = new LinearRegression() .setMaxIter(50) .setRegParam(0.01) .setElasticNetParam(0.8) val model = lr.fit(trainDF) val predictions = model.transform(trainDF) predictions.select("features", "label", "prediction").show(false)

线性回归在这里的含义很直白：把星期几和小时段作为 x，进站人数作为 y，模型学到的就是“工作日早八点的进站人数大概是多少”这类规律。RegParam 是正则项系数，设太小容易过拟合，设太大预测曲线会变成一条平线，0.01 通常是可以接受的起点。ElasticNetParam 设置为 0.8 表示偏向 L1 正则，特征少时效果更稳定。

真实地铁客流有明显的周期性，同一站点周一的早高峰曲线和周二差别不大，所以线性回归在“周几 + 时段”这两个特征下已经能跑出一个像样的趋势图。如果答辩时被问到预测精度，老实的回答是：基线模型看的是趋势和量级，不是精确值，想做准需要加节假日、天气、大型活动特征，这部分我在第 6 章展开。

4.4 结果写回与作业提交参数

聚合和预测结果不能只打在控制台上，要写回存储层供接口查询。常见做法是把结果存入 HBase 的另一张结果表，或者写到 MySQL。用 HBase 的代码如下：

import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put} import org.apache.hadoop.hbase.util.Bytes val conn: Connection = ConnectionFactory.createConnection(hbaseConf) val table = conn.getTable(TableName.valueOf("metro:flow_result")) hourlyFlow.collect().foreach { row => val put = new Put(Bytes.toBytes(s"${row.getString(0)}_${row.getString(2)}")) put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("total_inbound"), Bytes.toLong(row.getLong(3))) put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("total_outbound"), Bytes.toLong(row.getLong(4))) table.put(put) } table.close() conn.close()

注意这段代码是逐条 put，结果量大时性能很差。正确做法是先把 RDD 转成 HBase Put 的 RDD，再用saveAsNewAPIHadoopDataset批量写入。前者代码短，适合毕设演示；后者适合讲性能优化。我建议两个都写进报告，答辩时先演示能跑通的版本，再提一句生产环境用批量写入，姿态就站住了。

Spark 作业跑起来靠的是 spark-submit 提交，一个典型参数组合：

spark-submit \ --master yarn \ --deploy-mode cluster \ --executor-memory 4g \ --num-executors 4 \ --class com.metro.analysis.FlowHeatMapApp \ metro-flow-analysis-1.0.jar

executor-memory 和 num-executors 要根据集群资源调整，伪分布式环境下 2g 和 2 个 executor 就行，设太大会让 YARN 分配不到资源直接报错，这是新手最容易遇到的情况之一。

5. 部署与避坑：版本配套、启动顺序与高频踩坑记录

5.1 环境版本怎么选才不翻车

这套系统涉及的组件多，版本选错会引发一整串连锁报错。我在跑这套毕设源码时总结的稳定组合是：JDK 1.8 + Hadoop 2.10.x + HBase 1.4.x + Spark 2.4.x + Logstash 6.8.x。这个组合的核心逻辑是 Hadoop 2.x 配 HBase 1.4.x 是社区验证过的老搭档，Spark 2.4 对 HBase 1.4 的客户端兼容做得也比较好，而 Logstash 6.8 自带的 JRuby 版本安装 HBase 插件时不容易出编译错误。

如果你非要往上追新版本，比如 Spark 3.x 配 HBase 2.x，首先得改一堆 API 调用，其次 HBase 的 hbase-client jar 和 Spark 自带的 Hadoop client 会产生大量冲突类，调起来非常浪费时间。毕设的核心是跑通，不是追新。

5.2 启动顺序与首次验证路径

整个系统启动顺序有一条明确的依赖链，乱序启动会报各种连接异常。我的标准流程是：

# 1. 启动 HDFS start-dfs.sh # 2. 启动 HBase，等 HBase 的 16010 端口能访问再进下一步 start-hbase.sh # 3. 执行建表命令 hbase shell hbase.command # 4. 启动 Logstash，让数据开始流入 HBase logstash -f logstash-nginx.config # 5. 验证 HBase 里已经有数据 hbase shell > scan 'metro:passenger_flow', {LIMIT => 5} # 6. 提交 Spark 作业 spark-submit --master yarn --class com.metro.analysis.FlowHeatMapApp metro-flow-analysis-1.0.jar

第 5 步非常关键，很多人 Logstash 启动了但没有验证写入，直接跑 Spark 作业，结果自然是读不到数据，然后开始怀疑代码有 bug。正确的排查路径永远是先确认数据真的进了 HBase，再怀疑计算层。

5.3 高频踩坑记录

以下是跑这套源码过程中遇到的高频问题，每条都是现象、原因、解决方案三件套。

踩坑一：Logstash 启动时报 NoMethodError 或插件加载失败

现象：执行logstash -f logstash-nginx.config后，Logstash 报找不到 hbase 输出插件。原因：hbase 插件不是 Logstash 内置插件，需要手动安装，而且老版本 Logstash 的插件安装机制和新版本差异大。解决：在 Logstash 6.8 环境下执行bin/logstash-plugin install logstash-output-hbase，装完重启。如果安装失败，检查是不是用了 Logstash 7+，那个版本对 hbase 插件支持非常差。

踩坑二：Spark 作业报 ClassNotFoundException: org.apache.hadoop.hbase.client.ConnectionFactory

现象：Spark 跑起来没几秒就抛出这个类找不到。原因：Spark 作业的 classpath 里没有 HBase 客户端 jar，HBase 依赖没有打进 fat jar。解决：用maven-shade-plugin打包时把 hbase-client、hbase-common、hbase-server 打进去，同时把spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version设为 2，避免写 HDFS 时的小文件优化问题。

踩坑三：CSV 中文站名全部变成乱码

现象：HBase 里查出来的 station_name 是一串乱码。原因：CSV 文件是 GBK 编码，而 Logstash 的 csv filter 默认按 UTF-8 解析。解决：在 filter 区域加一行charset => "UTF-8"不行，需要先把 CSV 文件转成 UTF-8 无 BOM 格式，或者用fileinput 的charset => "GBK"配合。最简单的方法是直接用 VSCode 或 Notepad++ 把文件转成 UTF-8，一次到位。

踩坑四：Spark 聚合结果每天都是同一张表，没有按天分区

现象：运行两次作业，结果表里的旧数据和新数据混在一起，接口查询结果对不上。原因：结果表没有按日期做隔离，每次运行都往同一个 rowkey 上写，覆盖逻辑没写对。解决：rowkey 设计加日期前缀，比如2024-06-01_0101_07，每天的结果天然落在不同 rowkey 上，互不干扰。这个改动在 4.4 节的写入代码里只需要改一行字符串拼接。

踩坑五：线性回归预测值全部收敛到同一个数

现象：模型预测结果几乎是常数，完全看不出早晚高峰的波动。原因：特征太弱，只有 day_of_week 和 time_period，如果数据里恰好全是工作日，day_of_week 就失去了区分度，模型学到的是一条接近水平的线。解决：加入一小时前的进站量作为滞后特征，或者把 date 拆成“是否周末”“是否节假日”两个布尔字段，预测曲线立刻就有形状了。

提示：这五个坑几乎是每个跑这套系统的人都会遇到的，先把这五条记在心里，能省下好几个晚上的调试时间。

6. 二次开发思路：从离线统计升级到准实时与可视化

6.1 接 Kafka 把离线改成准实时

整套系统目前是“读 CSV 文件 → 批量计算 → 接口查结果”的离线链路，改造最自然的路径是引入 Kafka 做实时流。操作步骤如下：先在 Logstash input 里把 file 换成 kafka 插件，订阅地铁闸机实时上报的 topic，输出不变仍然写 HBase；再把 Spark 作业从离线 DataFrame 换成 Structured Streaming，用 5 分钟窗口做滚动聚合，结果写同一张结果表。改造完的架构就是准实时链路，延迟从小时级降到分钟级。

6.2 预测特征工程升级

第 4 章的线性回归模型预测结果比较平，二次开发的重点在特征工程。建议加入三类特征：一是时序滞后特征，取目标站前一天同时段的进站量作为输入；二是日历特征，标记当天是否工作日、是否节假日、是否大型活动日；三是空间特征，相邻站点的客流作为关联特征。特征列从 2 个扩到 6 到 8 个后，模型效果提升是最明显的，也方便写进论文作为创新点。

6.3 可视化对接与接口约定

结果要直观展示，最省力的做法是给 search.http 里的接口加一个前端页面，用 ECharts 折线图展示分时段客流，用地图散点图展示站点热度。接口返回 JSON 的格式建议保持统一：

{ "code": 0, "data": { "stationId": "0101", "date": "2024-06-01", "hourlyFlow": [ { "time_period": 7, "inbound": 1250, "outbound": 320 }, { "time_period": 8, "inbound": 2300, "outbound": 510 } ] } }

这个结构把每个小时的进出站数据放在数组里，前端拿到之后直接循环绘制，不需要后端再做二次加工。我在改这套代码的时候习惯性地把接口返回格式提前定死，前端和后端各写各的，联调时间能压缩一半。

这套源码我前前后后跑通了三遍，每次在环境配置上花的时间都比改代码多。第一遍老老实实按文档搭环境，第二遍自己换组件版本翻了一次车，第三遍才把所有版本问题摸透。从那以后我每拿到新的毕设源码，都会先看文件列表里有没有配置文件、有没有 .http 接口文件，有就先建表、导数据、调接口，把整条链路冒烟一遍，再去看核心计算代码。这套“先跑通、再读码、最后改”的顺序帮我避开过很多次自我怀疑，希望也能帮到你。

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