简介一份基于Spark的地铁大数据客流分析系统源码面向计算机科学与大数据相关专业的毕业设计学生也适合想快速上手Spark离线处理的开发者。项目覆盖数据采集、清洗、分析到可视化展示的完整链路可帮助理解Spark RDD与DataFrame等常用算子在地铁刷卡数据上的实际应用。压缩包共202个文件约42.77MB其中以Java与Scala源码为核心配以yaml、xml、properties等环境配置以及png图片和sql脚本便于还原项目结构与数据库表设计另有部分md说明文档辅助部署。目前已有327人浏览学习项目已经过本地编译验证下载后按配套说明配置环境即可运行整体难度适中能有效支撑毕业设计选题、代码参考与答辩讲解。1. 基于Spark的地铁大数据客流分析难在把海量刷卡记录变成调度指令地铁闸机的AFC系统一天会产生上千万条刷卡记录全网OD组合以十万计。客流分析要回答的其实只有三个问题某站在某个时段进出多少人、哪些区间最拥挤、换乘压力集中在哪。这些答案直接决定行车交路怎么排、限流措施几点启动。用Spark来做不是因为单机跑不动而是它能把清洗、OD聚合、断面推算放进同一条流水线从单机到集群迁移成本几乎为零。下面按架构选型、集群搭建、指标计算、落库可视化、答辩验证的顺序推进适合正在选型、或者代码写到一半不知道怎么组织模块和源码结构的人。2. 客流分析系统的选型依据与Spark集群搭建2.1 为什么选Spark而不是MapReduce或Flink选型先看数据形态。AFC原始数据是日增上千万行的结构化流水业务上要的是每天定时出结果属于典型的批处理场景。MapReduce能算但映射到OD矩阵和断面客流时每加一个指标就要重写一套Java MR程序调试一轮的代价足够写完三个DataFrame任务。Flink适合秒级延迟的实时限流预警但状态管理、watermark、checkpoint这些概念在毕设周期里会吃掉大量本应用于结果分析的时间而且答辩时很难一句话讲清状态后端。折中下来Spark是更稳妥的选项。它可以以Structured Streaming的形式做分钟级微批也可以用DataFrame批量算日结指标一套API覆盖两种模式。下面这张表是我在选型时会横向比的内容也是答辩为什么不用XX的直接依据。计算框架计算模型指标开发成本集群运维适合的客流场景MapReduce纯批量高每个指标都要写MR高周级或月级离线报表Spark微批批量低DataFrame或SQL均可中日级OD矩阵、断面客流、站点热度Flink真流式中状态管理复杂高秒级实时限流预警还有一个容易被忽略的点Spark的DataFrame在Catalyst优化器下会自动做谓词下推和列裁剪同样的聚合逻辑比手写RDD算子的代码短三分之一执行计划更稳定。毕设源码的可读性直接影响评分从RDD迁移到DataFrame这一步投入产出比很高。提示如果导师更看重实时性用Spark Structured Streaming可以边跑边答为什么不用Flink但核心指标仍然建议批量计算后落库实时链路只做增量刷新。2.2 从零搭起Spark集群本地模式与Standalone的安装与使用毕设阶段不推荐一上来就搭三台机器的YARN集群。大数据集群部署策略里有一条反复被验证的经验先在单机把任务跑通再谈分布式。本地模式适合写代码调试Standalone适合演示集群效果YARN留着等真有多个节点和HDFS需求时再上。下面的流程是spark的安装与使用里最常用的一套。# 选Spark 3.x的稳定版预编译包配置环境变量 export SPARK_HOME/opt/spark export PATH$SPARK_HOME/bin:$PATH # 单机Standalone把默认配置写进spark-defaults.conf cat $SPARK_HOME/conf/spark-defaults.conf EOF spark.master spark://localhost:7077 spark.executor.memory 2g spark.driver.memory 1g spark.sql.shuffle.partitions 8 EOF # 启动Master和Worker $SPARK_HOME/sbin/start-master.sh $SPARK_HOME/sbin/start-worker.sh spark://localhost:7077 # 验证进入spark-shell后能看到已连接的Worker $SPARK_HOME/bin/spark-shell --master spark://localhost:7077逐个说参数含义。spark.master指向Standalone的Master地址写成spark://格式可以在spark-submit和spark-shell里统一使用spark.executor.memory是Executor的堆内内存后面groupBy和join的数据都吃这块毕设机器内存16G的话给2g比较稳spark.sql.shuffle.partitions控制DataFrame触发shuffle时的分区数默认200单机小数据量会空转大量task改成8能明显加快。注意spark.executor.memory不要超过机器可用内存的三分之二否则Worker会频繁Full GC表现是任务卡在某个stage不结束日志里全是GC开销超限。2.3 客流数据的三种接入方式与选型判断数据接入层决定后面所有指标的口径。毕设里常见三种做法直接读CSV、JDBC读MySQL、Kafka接实时流。直接读CSV最省事AFC导出的刷卡流水本身就是CSV不用搭额外组件JDBC适合已经先做了一个管理前端、数据落在数据库里的情况Kafka适合演示大屏每分钟刷新的效果但需要同时维护生产者脚本和Streaming任务调试成本最高。接入方式数据形态演示效果实施成本直接读CSV文件日结指标完整最低JDBC读MySQL关系表与Web端共用数据源中Kafka Structured Streaming消息流大屏分钟级刷新高我一般会先用CSV把整条链路打通最后一周有余力再补Kafka。下面的CSV接入写法里schema必须要提前声明否则Spark会反复推断列类型每次跑任务多花几十秒。from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(metro-afc-etl) \ .master(local[4]) \ .config(spark.sql.shuffle.partitions, 8) \ .getOrCreate() afc_schema card_id STRING, line_id STRING, station_id STRING, in_time TIMESTAMP, out_time TIMESTAMP, in_station STRING, out_station STRING, fare DOUBLE df spark.read \ .option(header, true) \ .option(timestampFormat, yyyy-MM-dd HH:mm:ss) \ .schema(afc_schema) \ .csv(data/afc_2024_06_01.csv)master里写local[4]表示本机4个线程并行数据量小的时候比local[1]快4倍。timestampFormat必须和AFC导出的时间字符串完全一致解析不了的字段会变null这一步错了后面所有按时间的聚合都会缺数。3. 用Spark DataFrame完成客流ETL与OD、断面指标计算3.1 AFC原始数据的清洗规则与去重顺序地铁客流数据看着整齐实际脏数据集中在四个位置闸机重复刷卡、进站无出站、时间倒挂、站名字段混入全角空格。清洗顺序会影响指标结果所以先去重、再过滤、最后做字段归一顺序不要反过来。先去重能降低后续groupBy和join的数据量先过滤再去重可能把本应合并的重复记录误删。from pyspark.sql.functions import col, trim, regexp_replace # 1) 去重同一张卡同一进站时间视为重复刷卡 dedup df.dropDuplicates([card_id, in_time, out_time]) # 2) 过滤无出站记录或时间倒挂的数据不参与OD计算 valid dedup.filter( col(out_time).isNotNull() (col(out_time) col(in_time)) ) # 3) 归一全角空格和普通空格统一去掉避免同名站被当成两个站 clean valid.withColumn( in_station, trim(regexp_replace(col(in_station), |\\s, )) ).withColumn( out_station, trim(regexp_replace(col(out_station), |\\s, )) )dropDuplicates的subset要选够粒度只按card_id去重会把一个人一天内的多段行程全部干掉这是最常见的误用。时间倒挂的记录可以单独统计成异常刷卡口径与正常客流分开计数答辩时能讲清这个口径就是加分点。3.2 站点进出量、OD矩阵、断面客流三个指标的DataFrame写法三个核心指标的计算逻辑如下站点进出量按站时间窗聚合OD矩阵按进站出站时间窗聚合断面客流要把OD记录按线路展开成途经区间再聚合。前两个指标用groupBy就能完成断面客流需要一张线路站点顺序表配合UDF展开。from pyspark.sql.functions import window, count, col, explode, udf from pyspark.sql.types import ArrayType, StringType # 指标1每半小时各站进站量 station_hour clean.groupBy( window(col(in_time), 30 minutes), col(station_id) ).agg(count(card_id).alias(in_count)) # 指标2OD矩阵按小时聚合用于桑基图和流向分析 od_matrix clean.groupBy( col(in_station), col(out_station), window(col(in_time), 60 minutes) ).agg(count(card_id).alias(od_flow)) # 指标3断面客流用线路站点表把OD展开成途经站 line_stations { line_1: [station_001, station_002, station_003, station_004], line_2: [station_005, station_006, station_007], } def stations_between(in_st, out_st): try: for seq in line_stations.values(): if in_st in seq and out_st in seq: i, j seq.index(in_st), seq.index(out_st) return seq[i:j] if i j else seq[j:i] except (KeyError, ValueError): return None return None expand_udf udf(stations_between, ArrayType(StringType())) section_flow clean \ .withColumn(pass_stations, expand_udf(col(in_station), col(out_station))) \ .select(line_id, in_time, explode(col(pass_stations)).alias(pass_station)) \ .groupBy(line_id, pass_station, window(col(in_time), 30 minutes)) \ .agg(count(card_id).alias(section_flow))window函数里30 minutes是窗口大小早高峰分析用30分钟粒度合适换5 minutes数据量大5倍但能看出冲击细节。stations_between返回站名序列explode展开成一条记录一个途经站再做groupBy就得到断面流量。结果里station_001到station_002区间的客流量本质是所有经过该区间的OD记录数之和。UDF速度不算最快但毕设数据量下完全够用没必要为了性能引入复杂的高阶函数。3.3 防止OOM和倾斜spark内存与shuffle参数怎么设客流数据里有一个天然倾斜点早高峰的换乘大站。枢纽站进站量可能是普通站的三十倍groupBy时这一个key会把某个task拖到超时。Spark 3.x的AQE能自动处理一部分倾斜但需要先把开关打开。除此之外下面三个参数是DataFrame任务里最常调的。参数默认值建议值作用spark.sql.shuffle.partitions200核心数×2~4控制shuffle输出分区数过大会产生大量空taskspark.default.parallelism自动核心数×2~3影响stage初始并行度spark.sql.adaptive.enabledfalsetrue开启动态合并与倾斜join优化$SPARK_HOME/bin/spark-submit \ --master spark://localhost:7077 \ --executor-memory 2g \ --driver-memory 1g \ --conf spark.sql.shuffle.partitions8 \ --conf spark.default.parallelism8 \ --conf spark.sql.adaptive.enabledtrue \ metro_etl.py提交任务时我把参数全部写到spark-submit命令行而不是代码里换机器跑批不用改源码。任务跑挂时先分两类日志如果gc开销超过限制说明Executor内存不够往上加spark.executor.memory如果某个task反复失败且伴随FetchFailedException多半是数据倾斜先确认AQE已开启再检查执行计划里倾斜key的分区情况。spark内存相关的高频面试题集中在Executor堆内和堆外的划分答辩时能说出storage和execution共用一块内存池互相可以抢占这句基本就说明理解到位了。4. 客流分析结果落库到可视化大屏的完整链路4.1 Spark结果写回MySQL批量落库的参数控制DataFrame计算完不能直接给前端用需要落库。常见做法是Spark算完后写MySQLWeb后端只查MySQL这样演示时前端不会因为Spark任务重跑而白屏。写JDBC时必须控制batchsize一次性写入几十万行结果如果逐条插入现场演示会在这一步卡住。# 覆盖写OD结果表batchsize控制单批次行数 od_matrix.write \ .mode(overwrite) \ .option(batchsize, 500) \ .option(truncate, true) \ .jdbc( urljdbc:mysql://localhost:3306/metro ?useSSLfalserewriteBatchedStatementstrue, tableod_flow_result, properties{user: root, password: 123456} )rewriteBatchedStatementstrue是MySQL批量写入生效的关键不加这个参数batchsize不生效仍然逐条execute。mode选overwrite适合日结全量场景如果做增量改成append并配合日期分区字段防止同一天的数据写重。结果表的字段要和DataFrame列名一致否则JDBC映射报错提前用printSchema()核对一遍最省时间。4.2 用Spring Boot提供客流查询接口后端接口尽量薄只做查表→返回JSON不要在Controller里写聚合逻辑。聚合已经在Spark层做完了后端重复算一是慢二是容易和Spark口径不一致。下面的接口按站点和日期查分时进站量用JdbcTemplate直接查结果表。RestController RequestMapping(/api/flow) public class FlowController { private final JdbcTemplate jdbcTemplate; public FlowController(JdbcTemplate jdbcTemplate) { this.jdbcTemplate jdbcTemplate; } GetMapping(/station/hour) public ListMapString, Object stationHour( RequestParam String stationId, RequestParam String date) { String sql SELECT time_bucket, in_count FROM station_hour_flow WHERE station_id ? AND stat_date ? ORDER BY time_bucket ; return jdbcTemplate.queryForList(sql, stationId, date); } }参数用?占位符绑定避免拼接SQL引入注入风险答辩抽查时这里经常被问。station_id和stat_date要建联合索引否则大屏轮询时接口会随结果表变大而越来越慢。如果前端刷新频率到了每秒一次给接口加Spring Cache并配置5分钟过期能挡住大部分重复查询还要避免在循环里逐条查库那是典型的N1查询接口会被越刷越慢。4.3 数据大屏的三类图表与ECharts对接大屏不要贪多做三个能自圆其说的模块就够全网总览、分时趋势、OD流向。热力站点图在地图API可用时是加分项但地图服务的密钥配置和站点坐标数据准备会占大量时间不作为必做项。下面是大屏模块与后端接口的对应关系。大屏模块图表类型对应接口刷新方式全网进出站总览数字翻牌柱状图/api/flow/summary小时级分时进站趋势面积折线图/api/flow/station/hour分钟级OD客流量Top10桑基图/api/flow/od/top小时级前端用ECharts时关键是把后端返回的数组结构对齐到xAxis和series。下面这段是分时趋势图的加载逻辑轮询间隔和后端缓存过期时间对齐避免前端刷太快打满MySQL连接。async function loadStationTrend(stationId, date) { const resp await fetch( /api/flow/station/hour?stationId${stationId}date${date} ); const rows await resp.json(); chart.setOption({ xAxis: { type: category, data: rows.map(r r.time_bucket) }, yAxis: { type: value, name: 进站量 }, series: [{ type: line, smooth: true, areaStyle: {}, data: rows.map(r r.in_count) }] }); } // 大屏打开后每5分钟拉一次与后端缓存过期时间对齐 loadStationTrend(station_101, 2024-06-01); setInterval( () loadStationTrend(station_101, 2024-06-01), 5 * 60 * 1000 );如果演示时想做出实时感可以加一个只刷新最近5分钟进站量的轻接口而不是整屏重绘。把time_bucket排序后只取最后一条渲染增量视觉上就是大屏在跳动的实时客流。用React或Vue写大屏时接口结构不变只是把fetch逻辑封装进hooks或组合式函数里后端完全不用动。5. 答辩前必做的数据对账与Spark调优验证5.1 用固定种子生成可复现的模拟客流数据真实AFC数据涉及隐私演示时换模拟数据是常见做法。生成模拟数据最关键的细节是固定随机种子这样每次运行生成的CSV完全一致Spark结果、MySQL表、大屏截图三者能对上号。import random import pandas as pd random.seed(42) # 固定种子保证演示可复现 stations [fstation_{i:03d} for i in range(1, 21)] rows [] for _ in range(20000): in_st random.choice(stations) out_st random.choice([s for s in stations if s ! in_st]) h random.randint(6, 23) rows.append({ card_id: fC{random.randint(10000, 99999)}, in_time: f2024-06-01 {h:02d}:{random.randint(0, 59):02d}:00, out_time: f2024-06-01 {h 1:02d}:{random.randint(0, 59):02d}:00, in_station: in_st, out_station: out_st, fare: random.choice([3.0, 4.0, 5.0]) }) pd.DataFrame(rows).to_csv(demo_afc.csv, indexFalse)生成的字段与2.3节的schema对齐放进data目录就能直接跑通ETL。20万条数据本地生成只需几秒重跑代价低适合边调参边验证。5.2 用MySQL交叉验证Spark聚合结果答辩时结果对不对比怎么算更难答。交叉验证做法是抽一个时间窗加一两个站点把Spark输出表与对原始数据直接做的SQL统计对比。SELECT station_id, 08:00-08:30 AS time_bucket, COUNT(*) AS in_count FROM afc_raw WHERE station_id station_003 AND in_time 2024-06-01 08:00:00 AND in_time 2024-06-01 08:30:00 GROUP BY station_id;对比时两个口径必须对齐Spark侧先做了dropDuplicates和空值过滤MySQL侧也要先做同样的去重过滤否则两边数字必然对不上。对账通过后把SQL结果和Spark输出记录存成截图答辩现场直接展示比口头解释应该是对的有说服力得多。5.3 用checkpoint和结果表缓存兜底演示现场演示最容易翻车的是Spark任务中途失败要重跑。checkpoint可以把中间stage结果落到磁盘长链路跑到第三个stage挂掉时不用从头再算。调用方式很简单但位置有讲究spark.sparkContext.setCheckpointDir(file:///tmp/spark_checkpoint) od_matrix od_matrix.checkpoint() section_flow section_flow.checkpoint()注意checkpoint会切断血缘不要每加一步都调用只在链路中最长、最贵的那次shuffle之后checkpoint一次能显著缩短失败恢复时间。大屏侧依赖的是MySQL结果表而不是Spark里的DataFrame所以Spark重跑不影响前端展示这才是整套系统演示时最稳的兜底设计。本文还有配套的精品资源点击获取 基于Spark的地铁客流分析：从AFC数据到可视化大屏实战
简介一份基于Spark的地铁大数据客流分析系统源码面向计算机科学与大数据相关专业的毕业设计学生也适合想快速上手Spark离线处理的开发者。项目覆盖数据采集、清洗、分析到可视化展示的完整链路可帮助理解Spark RDD与DataFrame等常用算子在地铁刷卡数据上的实际应用。压缩包共202个文件约42.77MB其中以Java与Scala源码为核心配以yaml、xml、properties等环境配置以及png图片和sql脚本便于还原项目结构与数据库表设计另有部分md说明文档辅助部署。目前已有327人浏览学习项目已经过本地编译验证下载后按配套说明配置环境即可运行整体难度适中能有效支撑毕业设计选题、代码参考与答辩讲解。1. 基于Spark的地铁大数据客流分析难在把海量刷卡记录变成调度指令地铁闸机的AFC系统一天会产生上千万条刷卡记录全网OD组合以十万计。客流分析要回答的其实只有三个问题某站在某个时段进出多少人、哪些区间最拥挤、换乘压力集中在哪。这些答案直接决定行车交路怎么排、限流措施几点启动。用Spark来做不是因为单机跑不动而是它能把清洗、OD聚合、断面推算放进同一条流水线从单机到集群迁移成本几乎为零。下面按架构选型、集群搭建、指标计算、落库可视化、答辩验证的顺序推进适合正在选型、或者代码写到一半不知道怎么组织模块和源码结构的人。2. 客流分析系统的选型依据与Spark集群搭建2.1 为什么选Spark而不是MapReduce或Flink选型先看数据形态。AFC原始数据是日增上千万行的结构化流水业务上要的是每天定时出结果属于典型的批处理场景。MapReduce能算但映射到OD矩阵和断面客流时每加一个指标就要重写一套Java MR程序调试一轮的代价足够写完三个DataFrame任务。Flink适合秒级延迟的实时限流预警但状态管理、watermark、checkpoint这些概念在毕设周期里会吃掉大量本应用于结果分析的时间而且答辩时很难一句话讲清状态后端。折中下来Spark是更稳妥的选项。它可以以Structured Streaming的形式做分钟级微批也可以用DataFrame批量算日结指标一套API覆盖两种模式。下面这张表是我在选型时会横向比的内容也是答辩为什么不用XX的直接依据。计算框架计算模型指标开发成本集群运维适合的客流场景MapReduce纯批量高每个指标都要写MR高周级或月级离线报表Spark微批批量低DataFrame或SQL均可中日级OD矩阵、断面客流、站点热度Flink真流式中状态管理复杂高秒级实时限流预警还有一个容易被忽略的点Spark的DataFrame在Catalyst优化器下会自动做谓词下推和列裁剪同样的聚合逻辑比手写RDD算子的代码短三分之一执行计划更稳定。毕设源码的可读性直接影响评分从RDD迁移到DataFrame这一步投入产出比很高。提示如果导师更看重实时性用Spark Structured Streaming可以边跑边答为什么不用Flink但核心指标仍然建议批量计算后落库实时链路只做增量刷新。2.2 从零搭起Spark集群本地模式与Standalone的安装与使用毕设阶段不推荐一上来就搭三台机器的YARN集群。大数据集群部署策略里有一条反复被验证的经验先在单机把任务跑通再谈分布式。本地模式适合写代码调试Standalone适合演示集群效果YARN留着等真有多个节点和HDFS需求时再上。下面的流程是spark的安装与使用里最常用的一套。# 选Spark 3.x的稳定版预编译包配置环境变量 export SPARK_HOME/opt/spark export PATH$SPARK_HOME/bin:$PATH # 单机Standalone把默认配置写进spark-defaults.conf cat $SPARK_HOME/conf/spark-defaults.conf EOF spark.master spark://localhost:7077 spark.executor.memory 2g spark.driver.memory 1g spark.sql.shuffle.partitions 8 EOF # 启动Master和Worker $SPARK_HOME/sbin/start-master.sh $SPARK_HOME/sbin/start-worker.sh spark://localhost:7077 # 验证进入spark-shell后能看到已连接的Worker $SPARK_HOME/bin/spark-shell --master spark://localhost:7077逐个说参数含义。spark.master指向Standalone的Master地址写成spark://格式可以在spark-submit和spark-shell里统一使用spark.executor.memory是Executor的堆内内存后面groupBy和join的数据都吃这块毕设机器内存16G的话给2g比较稳spark.sql.shuffle.partitions控制DataFrame触发shuffle时的分区数默认200单机小数据量会空转大量task改成8能明显加快。注意spark.executor.memory不要超过机器可用内存的三分之二否则Worker会频繁Full GC表现是任务卡在某个stage不结束日志里全是GC开销超限。2.3 客流数据的三种接入方式与选型判断数据接入层决定后面所有指标的口径。毕设里常见三种做法直接读CSV、JDBC读MySQL、Kafka接实时流。直接读CSV最省事AFC导出的刷卡流水本身就是CSV不用搭额外组件JDBC适合已经先做了一个管理前端、数据落在数据库里的情况Kafka适合演示大屏每分钟刷新的效果但需要同时维护生产者脚本和Streaming任务调试成本最高。接入方式数据形态演示效果实施成本直接读CSV文件日结指标完整最低JDBC读MySQL关系表与Web端共用数据源中Kafka Structured Streaming消息流大屏分钟级刷新高我一般会先用CSV把整条链路打通最后一周有余力再补Kafka。下面的CSV接入写法里schema必须要提前声明否则Spark会反复推断列类型每次跑任务多花几十秒。from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(metro-afc-etl) \ .master(local[4]) \ .config(spark.sql.shuffle.partitions, 8) \ .getOrCreate() afc_schema card_id STRING, line_id STRING, station_id STRING, in_time TIMESTAMP, out_time TIMESTAMP, in_station STRING, out_station STRING, fare DOUBLE df spark.read \ .option(header, true) \ .option(timestampFormat, yyyy-MM-dd HH:mm:ss) \ .schema(afc_schema) \ .csv(data/afc_2024_06_01.csv)master里写local[4]表示本机4个线程并行数据量小的时候比local[1]快4倍。timestampFormat必须和AFC导出的时间字符串完全一致解析不了的字段会变null这一步错了后面所有按时间的聚合都会缺数。3. 用Spark DataFrame完成客流ETL与OD、断面指标计算3.1 AFC原始数据的清洗规则与去重顺序地铁客流数据看着整齐实际脏数据集中在四个位置闸机重复刷卡、进站无出站、时间倒挂、站名字段混入全角空格。清洗顺序会影响指标结果所以先去重、再过滤、最后做字段归一顺序不要反过来。先去重能降低后续groupBy和join的数据量先过滤再去重可能把本应合并的重复记录误删。from pyspark.sql.functions import col, trim, regexp_replace # 1) 去重同一张卡同一进站时间视为重复刷卡 dedup df.dropDuplicates([card_id, in_time, out_time]) # 2) 过滤无出站记录或时间倒挂的数据不参与OD计算 valid dedup.filter( col(out_time).isNotNull() (col(out_time) col(in_time)) ) # 3) 归一全角空格和普通空格统一去掉避免同名站被当成两个站 clean valid.withColumn( in_station, trim(regexp_replace(col(in_station), |\\s, )) ).withColumn( out_station, trim(regexp_replace(col(out_station), |\\s, )) )dropDuplicates的subset要选够粒度只按card_id去重会把一个人一天内的多段行程全部干掉这是最常见的误用。时间倒挂的记录可以单独统计成异常刷卡口径与正常客流分开计数答辩时能讲清这个口径就是加分点。3.2 站点进出量、OD矩阵、断面客流三个指标的DataFrame写法三个核心指标的计算逻辑如下站点进出量按站时间窗聚合OD矩阵按进站出站时间窗聚合断面客流要把OD记录按线路展开成途经区间再聚合。前两个指标用groupBy就能完成断面客流需要一张线路站点顺序表配合UDF展开。from pyspark.sql.functions import window, count, col, explode, udf from pyspark.sql.types import ArrayType, StringType # 指标1每半小时各站进站量 station_hour clean.groupBy( window(col(in_time), 30 minutes), col(station_id) ).agg(count(card_id).alias(in_count)) # 指标2OD矩阵按小时聚合用于桑基图和流向分析 od_matrix clean.groupBy( col(in_station), col(out_station), window(col(in_time), 60 minutes) ).agg(count(card_id).alias(od_flow)) # 指标3断面客流用线路站点表把OD展开成途经站 line_stations { line_1: [station_001, station_002, station_003, station_004], line_2: [station_005, station_006, station_007], } def stations_between(in_st, out_st): try: for seq in line_stations.values(): if in_st in seq and out_st in seq: i, j seq.index(in_st), seq.index(out_st) return seq[i:j] if i j else seq[j:i] except (KeyError, ValueError): return None return None expand_udf udf(stations_between, ArrayType(StringType())) section_flow clean \ .withColumn(pass_stations, expand_udf(col(in_station), col(out_station))) \ .select(line_id, in_time, explode(col(pass_stations)).alias(pass_station)) \ .groupBy(line_id, pass_station, window(col(in_time), 30 minutes)) \ .agg(count(card_id).alias(section_flow))window函数里30 minutes是窗口大小早高峰分析用30分钟粒度合适换5 minutes数据量大5倍但能看出冲击细节。stations_between返回站名序列explode展开成一条记录一个途经站再做groupBy就得到断面流量。结果里station_001到station_002区间的客流量本质是所有经过该区间的OD记录数之和。UDF速度不算最快但毕设数据量下完全够用没必要为了性能引入复杂的高阶函数。3.3 防止OOM和倾斜spark内存与shuffle参数怎么设客流数据里有一个天然倾斜点早高峰的换乘大站。枢纽站进站量可能是普通站的三十倍groupBy时这一个key会把某个task拖到超时。Spark 3.x的AQE能自动处理一部分倾斜但需要先把开关打开。除此之外下面三个参数是DataFrame任务里最常调的。参数默认值建议值作用spark.sql.shuffle.partitions200核心数×2~4控制shuffle输出分区数过大会产生大量空taskspark.default.parallelism自动核心数×2~3影响stage初始并行度spark.sql.adaptive.enabledfalsetrue开启动态合并与倾斜join优化$SPARK_HOME/bin/spark-submit \ --master spark://localhost:7077 \ --executor-memory 2g \ --driver-memory 1g \ --conf spark.sql.shuffle.partitions8 \ --conf spark.default.parallelism8 \ --conf spark.sql.adaptive.enabledtrue \ metro_etl.py提交任务时我把参数全部写到spark-submit命令行而不是代码里换机器跑批不用改源码。任务跑挂时先分两类日志如果gc开销超过限制说明Executor内存不够往上加spark.executor.memory如果某个task反复失败且伴随FetchFailedException多半是数据倾斜先确认AQE已开启再检查执行计划里倾斜key的分区情况。spark内存相关的高频面试题集中在Executor堆内和堆外的划分答辩时能说出storage和execution共用一块内存池互相可以抢占这句基本就说明理解到位了。4. 客流分析结果落库到可视化大屏的完整链路4.1 Spark结果写回MySQL批量落库的参数控制DataFrame计算完不能直接给前端用需要落库。常见做法是Spark算完后写MySQLWeb后端只查MySQL这样演示时前端不会因为Spark任务重跑而白屏。写JDBC时必须控制batchsize一次性写入几十万行结果如果逐条插入现场演示会在这一步卡住。# 覆盖写OD结果表batchsize控制单批次行数 od_matrix.write \ .mode(overwrite) \ .option(batchsize, 500) \ .option(truncate, true) \ .jdbc( urljdbc:mysql://localhost:3306/metro ?useSSLfalserewriteBatchedStatementstrue, tableod_flow_result, properties{user: root, password: 123456} )rewriteBatchedStatementstrue是MySQL批量写入生效的关键不加这个参数batchsize不生效仍然逐条execute。mode选overwrite适合日结全量场景如果做增量改成append并配合日期分区字段防止同一天的数据写重。结果表的字段要和DataFrame列名一致否则JDBC映射报错提前用printSchema()核对一遍最省时间。4.2 用Spring Boot提供客流查询接口后端接口尽量薄只做查表→返回JSON不要在Controller里写聚合逻辑。聚合已经在Spark层做完了后端重复算一是慢二是容易和Spark口径不一致。下面的接口按站点和日期查分时进站量用JdbcTemplate直接查结果表。RestController RequestMapping(/api/flow) public class FlowController { private final JdbcTemplate jdbcTemplate; public FlowController(JdbcTemplate jdbcTemplate) { this.jdbcTemplate jdbcTemplate; } GetMapping(/station/hour) public ListMapString, Object stationHour( RequestParam String stationId, RequestParam String date) { String sql SELECT time_bucket, in_count FROM station_hour_flow WHERE station_id ? AND stat_date ? ORDER BY time_bucket ; return jdbcTemplate.queryForList(sql, stationId, date); } }参数用?占位符绑定避免拼接SQL引入注入风险答辩抽查时这里经常被问。station_id和stat_date要建联合索引否则大屏轮询时接口会随结果表变大而越来越慢。如果前端刷新频率到了每秒一次给接口加Spring Cache并配置5分钟过期能挡住大部分重复查询还要避免在循环里逐条查库那是典型的N1查询接口会被越刷越慢。4.3 数据大屏的三类图表与ECharts对接大屏不要贪多做三个能自圆其说的模块就够全网总览、分时趋势、OD流向。热力站点图在地图API可用时是加分项但地图服务的密钥配置和站点坐标数据准备会占大量时间不作为必做项。下面是大屏模块与后端接口的对应关系。大屏模块图表类型对应接口刷新方式全网进出站总览数字翻牌柱状图/api/flow/summary小时级分时进站趋势面积折线图/api/flow/station/hour分钟级OD客流量Top10桑基图/api/flow/od/top小时级前端用ECharts时关键是把后端返回的数组结构对齐到xAxis和series。下面这段是分时趋势图的加载逻辑轮询间隔和后端缓存过期时间对齐避免前端刷太快打满MySQL连接。async function loadStationTrend(stationId, date) { const resp await fetch( /api/flow/station/hour?stationId${stationId}date${date} ); const rows await resp.json(); chart.setOption({ xAxis: { type: category, data: rows.map(r r.time_bucket) }, yAxis: { type: value, name: 进站量 }, series: [{ type: line, smooth: true, areaStyle: {}, data: rows.map(r r.in_count) }] }); } // 大屏打开后每5分钟拉一次与后端缓存过期时间对齐 loadStationTrend(station_101, 2024-06-01); setInterval( () loadStationTrend(station_101, 2024-06-01), 5 * 60 * 1000 );如果演示时想做出实时感可以加一个只刷新最近5分钟进站量的轻接口而不是整屏重绘。把time_bucket排序后只取最后一条渲染增量视觉上就是大屏在跳动的实时客流。用React或Vue写大屏时接口结构不变只是把fetch逻辑封装进hooks或组合式函数里后端完全不用动。5. 答辩前必做的数据对账与Spark调优验证5.1 用固定种子生成可复现的模拟客流数据真实AFC数据涉及隐私演示时换模拟数据是常见做法。生成模拟数据最关键的细节是固定随机种子这样每次运行生成的CSV完全一致Spark结果、MySQL表、大屏截图三者能对上号。import random import pandas as pd random.seed(42) # 固定种子保证演示可复现 stations [fstation_{i:03d} for i in range(1, 21)] rows [] for _ in range(20000): in_st random.choice(stations) out_st random.choice([s for s in stations if s ! in_st]) h random.randint(6, 23) rows.append({ card_id: fC{random.randint(10000, 99999)}, in_time: f2024-06-01 {h:02d}:{random.randint(0, 59):02d}:00, out_time: f2024-06-01 {h 1:02d}:{random.randint(0, 59):02d}:00, in_station: in_st, out_station: out_st, fare: random.choice([3.0, 4.0, 5.0]) }) pd.DataFrame(rows).to_csv(demo_afc.csv, indexFalse)生成的字段与2.3节的schema对齐放进data目录就能直接跑通ETL。20万条数据本地生成只需几秒重跑代价低适合边调参边验证。5.2 用MySQL交叉验证Spark聚合结果答辩时结果对不对比怎么算更难答。交叉验证做法是抽一个时间窗加一两个站点把Spark输出表与对原始数据直接做的SQL统计对比。SELECT station_id, 08:00-08:30 AS time_bucket, COUNT(*) AS in_count FROM afc_raw WHERE station_id station_003 AND in_time 2024-06-01 08:00:00 AND in_time 2024-06-01 08:30:00 GROUP BY station_id;对比时两个口径必须对齐Spark侧先做了dropDuplicates和空值过滤MySQL侧也要先做同样的去重过滤否则两边数字必然对不上。对账通过后把SQL结果和Spark输出记录存成截图答辩现场直接展示比口头解释应该是对的有说服力得多。5.3 用checkpoint和结果表缓存兜底演示现场演示最容易翻车的是Spark任务中途失败要重跑。checkpoint可以把中间stage结果落到磁盘长链路跑到第三个stage挂掉时不用从头再算。调用方式很简单但位置有讲究spark.sparkContext.setCheckpointDir(file:///tmp/spark_checkpoint) od_matrix od_matrix.checkpoint() section_flow section_flow.checkpoint()注意checkpoint会切断血缘不要每加一步都调用只在链路中最长、最贵的那次shuffle之后checkpoint一次能显著缩短失败恢复时间。大屏侧依赖的是MySQL结果表而不是Spark里的DataFrame所以Spark重跑不影响前端展示这才是整套系统演示时最稳的兜底设计。本文还有配套的精品资源点击获取