基于Spark与HBase的地铁大数据客流分析系统设计

简介这是一份基于Spark的地铁大数据客流分析系统毕业设计项目面向计算机相关专业学生及大数据入门学习者可支撑课程设计、毕业设计或自学实战。项目聚焦地铁客流的统计、预测与优化覆盖数据采集、清洗、分析、存储与可视化等完整链路。压缩包共194个文件约42.6MB包含Java/Scala源码、Spark作业脚本、SQL数据库脚本、XML/YAML配置、PNG截图等并有HBase脚本、日志采集配置及地铁客流CSV样例数据已有259人学习浏览。通过学习可获得完整项目参考包括需求分析、系统设计、测试数据与实验报告还配有答辩演示文稿与报告模板可掌握Spark SQL、Spark Streaming、MLlib等核心组件在真实场景中的用法并了解MySQL或HDFS等数据库的整合实践。适合希望从零搭建大数据分析系统、撰写毕设文档或提升工程能力的读者。1. 地铁客流分析不是写个报表而是一套完整的数据工程问题做地铁大数据客流分析系统这个毕设题目的同学最容易踩的第一个坑是把它当成一个“查数据库然后画图表”的普通管理系统来做。但你一旦拿到真实的地铁刷卡数据或者 AFC 闸机日志就会发现数据量根本不是 MySQL 单机能扛住的按线路、站点、时段、OD 去聚合SQL 跑一次要几分钟导师问一句“能不能近实时看今天早高峰的断面客流”整个方案就站不住了。这个毕设的核心矛盾不是“怎么展示数据”而是“海量明细数据怎么存、怎么算、怎么在秒级出结果”。本文要拆解的这套基于 Spark 的地铁大数据客流分析系统就是围绕这个矛盾来设计的完整方案用 Spark 做分布式计算引擎用 HBase 做列式存储用 Logstash 做数据采集管道再配一套数据库脚本和样例 CSV 数据覆盖从数据接入、清洗、聚合到趋势预测的完整链路。适合正在做大数据方向毕设、需要一套能跑通且能讲清楚架构的同学也适合想参考真实工程组织方式的从业者。2. 技术选型与数据链路为什么是 Spark、HBase 和 Logstash 的组合2.1 数据从哪来从 HTTP 接口抓取到 CSV 落地再到 Logstash 管道先看这套资源里实际存在的文件就能反推它的数据链路设计。压缩包里有一个szt-api.http和一个search.http这两个是 HTTP 请求脚本用 IDE 的 REST Client 插件就能直接执行通常用来从地铁客流服务接口或者搜索引擎接口拉取原始数据szmc.net-metro.csv是一份已经抓取落地的样例数据约定了数据格式logstash-nginx.config是 Logstash 的采集配置文件但它采集的不是一般网站日志而是 nginx 网关转发 API 请求时产生的访问日志。整条链路是HTTP 接口 → 响应数据转成 CSV 或打入日志 → Logstash 读取并清洗 → 写入下游存储。这个设计的合理性在于地铁客流数据的真实来源往往是多个分散的接口有的返回 JSON有的返回 CSV 文件还有的是通过反向代理暴露的查询服务。直接在代码里写死接口地址数据格式一变就要改源码非常被动。把 Logstash 放在中间做一层管道等于给数据接入装了一个“后悔药”接口返回格式变化时只需要改 Logstash 的 filter 配置Spark 作业和存储层完全不用动。input { file { path [/data/metro-logs/*.log] start_position beginning sincedb_path /dev/null codec json } } filter { mutate { convert { station_id integer } convert { passenger_count integer } } date { match [timestamp, yyyy-MM-dd HH:mm:ss] target timestamp } } output { hbase { zk_quorum node01:2181,node02:2181,node03:2181 table metro_passenger_flow columns [ info:station_id, info:line_id, info:passenger_count, info:ts ] } }这段配置干了三件事监听指定目录下的日志文件把 JSON 格式的日志解析成结构化数据把station_id和passenger_count从字符串转成数值类型这一步很关键否则后面 Spark SQL 做聚合时会因为类型不匹配直接报错最后把每条记录写入 HBase 表的info列族。sincedb_path设为/dev/null是调试阶段常用的做法意思是每次重启 Logstash 都从文件头重新读方便反复验证配置效果生产环境要去掉这个配置。2.2 存储选型HBase 为什么比 MySQL 更适配客流轨迹数据客流明细数据的特征是写入量大、按时间和站点维度查询多、更新极少、字段相对固定。这种模式正好落在 HBase 的擅长区间。HBase 是列式存储rowkey 设计好之后按地铁线路和时间的范围扫描极其高效。比如设计 rowkey 为线路ID_日期_站点ID查询“一号线某天所有站点的客流”就变成一次连续的范围扫描而不需要像 MySQL 那样做多表 JOIN。项目里有一个hbase.command文件里面装的就是建表和预分区命令。预分区是 HBase 使用中必须做的一步否则默认只有一个 region数据一多就会发生 region 分裂写入性能会明显抖动。create metro_passenger_flow, {NAME info, VERSIONS 1, COMPRESSION SNAPPY}, {SPLITS [01|, 02|, 03|, 04|, 05|]}这里的SPLITS是按线路编号的前缀做预分区01|到05|对应五条不同的地铁线路前缀。COMPRESSION SNAPPY是压缩算法选择Snappy 的压缩比和 CPU 开销在日志类数据上比较均衡。注意这里使用了分隔符|是为了避免线路编号出现前缀冲突比如1和10如果不用分隔符01|的 range 扫描会误伤其他线路的数据。2.3 计算引擎Spark 的三大组件分别解决什么问题这个系统在计算层用了 Spark 的多个组件对应不同场景。Spark SQL 用于离线批处理处理szmc.net-metro.csv或者 HBase 里的历史数据做按小时、按线路、按站点的聚合统计生成日报表和周趋势Spark Streaming 用于处理 Logstash 实时写入的日志数据统计当前时段进出站人数用来支撑“晚高峰拥挤度”这类近实时指标MLlib 则是用历史客流序列训练预测模型预测未来一小时某个站点的客流峰值。这三块正好对应毕设答辩时导师最爱问的三个问题数据怎么批量处理、怎么实时处理、有没有用机器学习。一套方案全部覆盖不需要额外堆砌技术栈。3. 总体架构设计四层架构与模块职责划分3.1 分层架构接入层、存储层、计算层、应用层整套系统的架构可以看作四层。接入层由 HTTP 请求脚本和 Logstash 组成负责把外部数据拉回来并做初步清洗存储层使用 HBase 存储明细数据使用本地文件系统或 HDFS 存储 CSV 原始文件和 Spark 作业的中间结果计算层是 Spark 集群运行三个相互独立的作业模块应用层则是查询接口和结果展示通过搜索接口把聚合结果暴露给前端或报表工具。层与层之间通过数据格式解耦接入层输出 JSON 或 CSV存储层以 HBase 的单元格形式保存计算层读出来转成 DataFrame应用层再查结果表。每一层只依赖下一层提供的标准接口替换任何一层都不会影响其他层。这个设计在毕设答辩时非常加分因为它体现了软件工程的分层思想而不是把代码全堆在一个主类里。3.2 数据库设计与 rowkey 规划数据库设计是本项目最需要提前想清楚的部分。HBase 没有传统表结构的约束所以“设计表”实际上是在设计 rowkey 和列族。我的建议是把表拆成两张一张存原始明细另一张存聚合结果。明细表的 rowkey 设计为lineId_date_time_stationId这样同一个线路同一天的数据在物理上连续存储Spark 扫描时能利用 HBase 的 region 局部性聚合结果表存储 Spark 计算后的统计数据供上层 API 直接查询不需要每次查询都跑一遍完整聚合。create metro_flow_daily, {NAME agg, VERSIONS 1}这张metro_flow_daily表的 rowkey 建议设计为date_lineId_stationId列族agg下存放total_in、total_out、peak_hour等预聚合字段。把聚合结果单独存一份等于用存储空间换查询时间这是大数据项目里非常典型的空间换时间策略也是后续做实时展示性能达标的关键。3.3 工程组织从配置文件到作业脚本的部署结构一个合格的 Spark 毕设项目代码结构一定不是平铺的。通常会分成四个目录etl放数据接入和清洗代码analysis放统计分析代码ml放预测模型代码resources放配置文件。压缩包里的.editorconfig、.gitignore说明项目做过工程化约束.gitignore里应该会忽略掉target/、*.class、*.jar这类构建产物。logstash-nginx.config属于部署配置文件hbase.command属于初始化脚本。这种组织方式的好处是任何人都能按目录名快速定位到对应功能模块导师检查代码时不需要你从头讲解。4. 核心模块实现从 Spark SQL 聚合到 MLlib 客流预测4.1 Spark SQL 离线统计按线路、站点、时段聚合客流离线统计模块解决的是“昨天全天各站进出站量是多少”“早高峰集中在哪些站”这类问题。核心逻辑是从 HBase 读取明细数据转成 DataFrame然后按维度做 groupBy 聚合。下面是一段可以直接跑的 Scala 代码骨架对应分析模块里的核心逻辑。import org.apache.spark.sql.SparkSession import org.apache.spark.sql.functions._ val spark SparkSession.builder() .appName(MetroDailyStat) .enableHiveSupport() .getOrCreate() // 从 HBase 读取明细数据通过 Catalog 映射表结构 val df spark.read .format(org.apache.hadoop.hbase.spark) .options(Map( hbase.zookeeper.quorum - node01:2181,node02:2181,node03:2181, catalog - {table:{namespace:default,name:metro_passenger_flow}, rowkey:key,columns:{ col0:{cf:rowkey,col:key,type:string}, line_id:{cf:info,col:line_id,type:string}, station_id:{cf:info,col:station_id,type:string}, passenger_count:{cf:info,col:passenger_count,type:int}, ts:{cf:info,col:ts,type:string} }}.stripMargin )) .load() // 解析时间字段提取小时维度 val result df .withColumn(hour, hour(to_timestamp(col(ts), yyyy-MM-dd HH:mm:ss))) .groupBy(line_id, station_id, hour) .agg( sum(passenger_count).as(total_count), count(*).as(record_count) ) .orderBy(line_id, station_id, hour) result.write.format(parquet).mode(overwrite) .save(/data/result/daily_stat)这段代码的关键点有三个catalog参数必须和 HBase 表结构完全对上列族的字段名写错一个就会在运行时抛NoSuchColumnExceptionhour(to_timestamp(...))的格式串必须与源数据的时间格式一致否则解析出来全是 null聚合结果会变成空表结果写入 Parquet 格式做列式存储后续查询的效率远高于直接存 CSV。写完聚合逻辑后建议先用df.createOrReplaceTempView(tmp)注册成临时表再用 Spark SQL 里的SELECT ... WHERE line_id 01做一次过滤验证确认结果字段语义正确再落盘。4.2 Spark Streaming 实时统计窗口内的进出站人数实时模块解决的是“当前这个时段各站进出站人数”的问题。Logstash 把数据写入 HBase 的同时Spark Streaming 用固定时间间隔拉取增量数据在窗口内做累加。这里有一个容易理解错的地方Spark Streaming 的窗口计算是基于微批的不是真正毫秒级实时但在毕设场景下秒级的延迟已经完全够用。import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.kafka010._ val ssc new StreamingContext(spark.sparkContext, Seconds(5)) // 假设 Logstash 采集的数据先进入 Kafka topic: metro-flow val kafkaParams Map( bootstrap.servers - node01:9092,node02:9092,node03:9092, key.deserializer - org.apache.kafka.common.serialization.StringDeserializer, value.deserializer - org.apache.kafka.common.serialization.StringDeserializer, group.id - metro-streaming-group, auto.offset.reset - latest ) val stream KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](Array(metro-flow), kafkaParams) ) stream.map(_.value()) .map(record { // 这里直接解析 JSON 数据 val fields record.split(,) (fields(0), fields(2).toInt) // (station_id, passenger_count) }) .reduceByKeyAndWindow(_ _, Seconds(60), Seconds(10)) .foreachRDD(rdd { rdd.foreachPartition(iter { // 把窗口结果写入 Redis 或结果表供前端实时刷新 }) }) ssc.start() ssc.awaitTermination()窗口参数Seconds(60), Seconds(10)的含义是每 10 秒计算一次最近 60 秒的数据。这个设计比按微批直接累加更符合“客流趋势”的语义因为 60 秒的窗口能平滑掉偶发的闸机抖动数据比如某台闸机临时故障导致的一分钟刷卡量异常偏高。写入结果时通常的做法是写 Redis 的 Hash 结构以station_id为 key、窗口结束时间为 field这样前端轮询时只需要一次HGETALL就能拿到全站实时客流。PreferConsistent策略会把 Kafka 分区尽量均匀分配给各 Spark Executor避免数据倾斜导致某个节点成为瓶颈。4.3 MLlib 客流预测用历史序列预测未来一小时客流量预测模块是答辩时最容易被追问的部分。这里不推荐用过于复杂的深度网络用 MLlib 里的线性回归或者决策树就足够重点是讲清楚训练数据的构造方式。预测目标不是某个具体的“下午三点客流量”而是一个滑动窗口模型用前 7 天同一时段、同一站点的客流数据加上当天是工作日还是节假日这个特征预测下一个小时的客流值。import org.apache.spark.ml.feature.VectorAssembler import org.apache.spark.ml.regression.LinearRegression // 特征列历史同时段客流、前一时段客流、是否为节假日、是否为高峰 val featureCols Array(hour_avg_7d, prev_1h_count, is_holiday, is_peak) val assembler new VectorAssembler() .setInputCols(featureCols) .setOutputCol(features) val lr new LinearRegression() .setMaxIter(50) .setRegParam(0.01) .setElasticNetParam(0.8) val pipeline new Pipeline().setStages(Array(assembler, lr)) val model pipeline.fit(trainDf) // 输出预测结果计算误差指标 val evaluateDf model.transform(testDf) evaluateDf.select(prediction, actual_count)hour_avg_7d这个特征是把过去 7 天的同一小时客流取平均值用于捕捉周期性规律prev_1h_count用前一小时的客流值捕捉短期趋势比如早高峰的爬坡阶段is_holiday用 0/1 标记捕捉节假日和工作日的巨大客流差异。训练之前一定要先把标签列actual_count和特征列分开否则会用答案去预测答案模型精度虚高但实际无效。评价指标上看 R² 或者 RMSE 两个值就够不要堆一堆指标增加答辩负担。一个合格模型的 R² 在 0.85 左右就是比较理想的水平低于 0.7 就要检查特征构造是否有明显遗漏。5. 常见问题排查部署与运行中最容易翻车的五个点5.1 Spark 作业提交后一直卡在 ACCEPTED 状态现象使用spark-submit提交作业后YARN 界面显示状态一直是 ACCEPTED日志不输出整个任务像卡死了一样。原因绝大多数情况是 Executor 申请的内存超过了集群可用资源。比如spark.executor.memory配了 4G但每个节点可用内存只有 3GResourceManager 无法满足申请条件任务就一直排队等待。解决先执行free -h查看每台节点的可用内存把spark.executor.memory和spark.executor.cores调低或者减少spark.executor.instances数量。较稳妥的做法是从最小配置开始验证1 个 executor、1G 内存、1 核跑通后再逐步增加资源。5.2 Logstash 能启动但 HBase 里查不到数据现象Logstash 进程正常运行日志也没有报错但用scan metro_passenger_flow查不到任何数据。原因Logstash 的 HBase output 插件与 HBase 服务端的版本不兼容最典型的是 HBase 开启了 Kerberos 认证但 Logstash 配置里没带 principal数据被认证层拦截了另一种情况是写入的 rowkey 重复后写入的数据被覆盖。解决先看 Logstash 日志里有没有org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException有就是服务端连接问题。如果确认版本兼容用hbase shell手动 put 一条数据测试表是否可写排除输出配置的错误。5.3 Spark SQL 读取 HBase 时报列族找不到现象代码和 HBase 表都存在但运行时抛出column family not found异常。原因catalog参数里的cf值写错了大小写或者名称HBase 是大小写敏感的INFO和info是两个完全不同的列族。还有一个常见坑是列名和 rowkey 用了同一个字段名导致行键列和普通列在映射时冲突。解决在 HBase shell 里执行describe metro_passenger_flow拿到准确的列族名再回头改catalog。rowkey 字段应该单独映射为cf: rowkey与 info 列族里的字段名避免重名。5.4 MLlib 训练时特征列包含空值导致程序崩溃现象模型训练在 fit 阶段报错提示输入特征包含 NaN 或 null。原因原始 CSV 数据里某些时段没有客流记录Spark 读进来以后这些行的特征列是空值LinearRegression 不允许特征中存在缺失值。解决在构造特征之前加一步df.na.fill(0, Seq(prev_1h_count))把缺失的历史客流值填成 0这是一种简化处理。更好的做法是先用groupBy station_id计算每个站点的平均客流填充因为不同站点的量级差别很大全填 0 会导致小客流站点的预测值被低估。5.5 实时模块延迟越来越大最终数据堆积现象程序运行 30 分钟后处理延迟从最初 5 秒增长到几十秒Kafka 里的 lag 持续增加。原因窗口计算里的foreachRDD内做了同步的批量写入操作比如每次把全量窗口数据写入 HBase这个写操作耗时过长拖慢了整个微批的处理节奏。解决把每次窗口计算的结果先写入内存或本地文件用单独的线程异步刷入 HBase。如果不能改代码就提高 Spark Streaming 的spark.streaming.kafka.maxRatePerPartition上限配合增加分区数来提升吞吐。最重要的是设置spark.streaming.stopGracefullyOnShutdown为 true否则直接 kill 进程会发生数据丢失或重复消费。6. 验收与答辩怎么证明这套系统真的能跑验收不是把界面截图贴进报告就完了而是要把“系统能用”变成可验证的过程。我的做法是准备一份验收清单按数据接入、离线统计、实时计算、预测四个模块逐项打勾。每项要有输入样例和预期输出比如输入szmc.net-metro.csv里某站点连续 5 天的数据然后断言daily_stat结果表里该站点的总客流等于 CSV 里所有记录之和。数值对不上但差值不大优先查数据清洗逻辑里有没有过滤掉异常值差值很大则优先查聚合的时间维度是不是发生了时区偏移日志类数据最容易出现 UTC 和本地时间混用的问题。答辩演示的顺序也建议固定下来先跑一次spark-submit提交离线统计作业同时展示作业执行进度日志拿到结果后到 HBase 里scan出数据接着启动实时模块手动向 Kafka 生产几条带当前时间戳的模拟客流数据观察窗口输出最后加载训练好的 MLlib 模型对测试数据集做一次预测并展示 RMSE 数值。整个演示控制在 10 分钟以内核心是让导师看到“数据能进去、算得出结果、结果能查出来”这条链路是通的。导师问“实时性和离线有什么区别”这类问题时直接回答 Spark Streaming 的微批机制和 Spark SQL 的批处理差异不要试图把 Spark Streaming 包装成毫秒级实时。遇到“为什么不用 Flink”的追问就承认 Flink 在真正毫秒级实时上更优但当前场景对延迟要求不高Spark 一套技术栈覆盖离线、实时和机器学习工程复杂度更低。诚实说明技术选型的权衡比强行辩护更能得分。我从那次之后每次交付项目都强制走一遍“数据全链路跑测 数值一致性校验 演示脚本演练”这三步能挡住八成以上的翻车现场。希望帮到你。本文还有配套的精品资源点击获取