基于Spark的地铁客流分析系统：架构、实践与避坑指南-开发者社区

简介：面向计算机专业毕业设计的完整项目，基于Spark的地铁大数据客流分析系统，以城市地铁客流数据为分析对象，覆盖数据采集、清洗、存储、分析、可视化与客流预测等环节，适合大数据方向学生用于课程设计、毕设参考或技术实战练习。压缩包共194个文件，包含Java/Scala源码、Spark作业脚本、SQL/HBase数据库脚本、XML/YAML配置、CSV测试数据，以及项目PPT、PNG图像和日志文件等，整体42.6MB，按模块组织、目录清晰。已有259人学习/下载。通过系统可掌握Spark SQL、Streaming、MLlib等核心组件的应用，理解客流统计、时间序列分析与结果展示思路；文档部分涵盖需求分析、系统设计、数据库设计等软件工程材料，日志配置、接口调用记录与Shell脚本则有助于环境搭建和排错，方便完整复现毕业设计全过程并开展二次开发。

1. 基于Spark的地铁客流分析毕设：为什么值得拆开看

地铁客流数据有三个和大数据场景天然贴合的特征：全天候产生、时间连续、空间分布不均。高峰期一个站点的十分钟客流量能抵低谷期一天的量，这种数据跑在单机MySQL上做一次小时级窗口聚合都捉襟见肘。基于Spark的地铁大数据客流分析系统，就是用Spark分布式计算来解决这个瓶颈，覆盖客流清洗、统计、预测的完整链路：Logstash采集、HBase存储、Spark计算，附带CSV数据集与HTTP接口文件。

适合两类人：一类是准备大数据方向毕设的学生，缺一个能讲清架构、能真正跑通的参考项目；另一类是工作里需要快速上手Spark的从业者，想拿真实数据练手而不是反复跑WordCount。这份压缩包拆开是源码和配置，合起来就是一条可复现的地铁客流分析Pipeline。

2. 系统架构拆解：HBase、Logstash与Spark是谁在解决什么问题

2.1 为什么不是MySQL：HBase加Spark的底层逻辑

这个项目选型不是拍脑袋，从数据规模与写入模式上都能站住脚。

城市地铁的客流数据来源与业务人员手工录入完全不同，它由闸机或AFC自动售检票系统持续产生事件流。假设一个城市有200个站，每分钟产出上千条进出站事件，一天的原始记录就接近百万级。这种高并发写入场景，MySQL需要分库分表、读写分离才能撑住，而这些方案都会显著增加开发量。HBase做列族存储，按RowKey有序排列，高吞吐写入是它的主场，还天然支持按时间范围做高效Scan。

第二，数据落库后的分析需求通常是多维聚合：按线路、站点、小时、上下行分组统计客流量，或者对比工作日与节假日的曲线差异。HBase擅长点查和范围扫，但复杂聚合不是它的强项。Spark SQL则专门做这件事——把HBase某个时间段的数据扫出来映射成DataFrame，再用GroupBy、Join、窗口函数做分布式聚合，计算压力被分散到集群多个executor上。一个扮演存储角色，一个扮演计算角色，分工明确。

第三，从毕设答辩视角看，HBase加Spark能在架构图上画出完整闭环：API采集、Logstash清洗、HBase落地、Spark分析、结果输出。如果选MySQL方案，这个闭环会断在“大数据量”这个卖点上，评审更容易追问“为什么不用Excel”。选择HBase+Spark，至少可以在数据规模和计算模式上给出有说服力的答案。

资源包里hbase.command脚本就是数据接入层的操作证明，用HBase Shell命令建表、插数、Scan验证。我见过不少毕设项目把精力放在Spark代码上，数据链路反而一塌糊涂，最后Spark和HBase没连上。这个脚本的存在说明项目作者在数据接入上是有实际操作的。

2.2 hbase.command与logstash-nginx.config：数据接入层的真实形态

hbase.command通常是一组HBase Shell命令，用来初始化表结构和验证数据。常见内容与关键参数如下：

# 创建客流原始数据表，列簇为info，不保留历史版本 create 'metro_flow', {NAME => 'info', VERSIONS => 1} # 预分区：按线路前缀拆4个region，避免写入热点 create 'metro_flow', {NAME => 'info'}, {SPLITS => ['01', '02', '03', '04']} # 查看表是否存在 list 'metro_flow' # 插入一条示例数据（RowKey = 线路站点编码 + 时间戳） put 'metro_flow', '01001_20240101120000', 'info:station', '站厅A' put 'metro_flow', '01001_20240101120000', 'info:in_count', '120' # 按时间范围Scan，验证数据落库 scan 'metro_flow', {STARTROW => '01001_20240101', ENDROW => '01001_20240102'}

这里三个参数是真正影响性能的地方。

第一是RowKey设计。把线路站点编码放前、时间戳放后，同一站点的数据在物理存储上相邻，Scan就从随机读变成顺序读。如果反过来写成时间戳在前，同一时间点所有站点的数据散落在不同区域，查询时HBase要做大量随机IO。RowKey设计一旦定下来后期很难改，相当于一把后悔药都没有。

第二是SPLITS预分区。不预分区的表只有一个region，所有写入都压在同一个RegionServer上，集群再有5台机器也只有1台在干活。按线路前缀拆成多个region后，写入可以分流。分区边界要根据实际RowKey前缀来定，不要把四个分区边界设成无意义的数字。

第三是VERSIONS参数。客流数据是典型的append-only流式数据，旧版本没有保留价值，设成1节省存储。代价是无法用旧版本做回滚分析，但客流场景不需要。

接着是logstash-nginx.config。在真实地铁系统里，客流数据通常不是直接写HBase，而是由Nginx代理层接收请求后输出access log，Logstash再按固定间隔读取增量日志、解析成结构化字段、写出到HBase。配置分input、filter、output三段，常见写法如下：

input { file { path => "/var/log/nginx/metro-access.log" start_position => "beginning" sincedb_path => "/dev/null" codec => "json" } } filter { grok { match => { "message" => "%{TIMESTAMP_ISO8601:ts} %{DATA:station} %{WORD:direction} %{INT:flow}" } } date { match => ["ts", "ISO8601"] target => "@timestamp" } } output { hbase { table => "metro_flow" rowkey => "%{station}_%{@timestamp}" columns => [ "info:direction", "%{direction}", "info:flow", "%{flow}" ] } }

这里有一个容易被忽略却影响巨大的配置：sincedb_path。它记录Logstash读取文件的offset位置。设置成/dev/null时，每次重启Logstash都会从文件开头重新读一遍，开发调试期方便，但生产环境一旦重启，全部历史日志会重复灌入HBase，数据直接翻倍。初跑通后务必改成持久化路径，比如sincedb_path => "/var/lib/logstash/sincedb/metro.db"。

grok插件是Logstash解析文本日志的常用方式，用正则按命名捕获字段。如果Nginx日志格式调整过但grok没同步更新，字段会解析失败，output段引用变量变成空。排查方法是先跑logstash -f 配置文件 --config.test_and_exit验证语法，再用stdout输出插件代替hbase输出，先确认解析结构无误再接HBase。

2.3 szmc.net-metro.csv：字段结构与分析切入点

szmc.net-metro.csv，szmc大概率是深圳地铁的拼音缩写，是整个项目的核心数据集。毕设里的CSV规模一般在几十MB到几百MB，字段设计参考了真实AFC系统的数据结构。常见字段大致如下：

字段名	类型	含义
station_code	string	站点编码
station_name	string	站点名称
line_no	int	线路编号
direction	string	上行/下行
ticket_type	string	单程票/卡/扫码
in_time	timestamp	进站时间
out_time	timestamp	出站时间
flow_count	int	客流量

拿到CSV后第一件事不是直接写Spark代码，而是先确认文件本身的形态。用命令行看一眼：

# 看文件编码，避免UTF-8 BOM导致表头污染 file szmc.net-metro.csv # 看每行长度，排除跨行字段 awk '{print NR, length($0)}' szmc.net-metro.csv | head -5 # 看表头和前两行内容 head -3 szmc.net-metro.csv

如果发现表头第一列有看不见的BOM字符，用sed去BOM再继续。字段时间格式不统一也是常见问题，有的行是2024-01-01 12:00:00，有的行是2024/01/01 12:00:00，这种情况Spark的TimestampType解析会部分失败。可以先用sed或Excel批量统一格式，比在Spark里写复杂解析逻辑省事得多。

分析切入点的选择直接影响毕设功能列表是否饱满，我建议覆盖三个方向。一是分时段客流统计，找出早高峰和晚高峰的时间窗口，用Spark SQL的hour函数加groupBy就能实现。二是站点热度排行，算每个站的日进出站总量，输出Top N。三是线路断面客流，识别最拥挤的区间，这个方向需要把站点线间距数据join进来，是展示多表join能力的好素材。三个方向代码量都不大，但组合起来能覆盖客流分析系统的常见功能，答辩时也容易讲清楚设计思路。

3. Spark集群搭建与代码落地：从local模式到spark-submit的完整链路

3.1 Spark的安装与使用：集群搭建的两种模式怎么选

先明确一个原则：这份毕设数据量不大，如果本机内存只有8G，不要一上来搭三节点集群。先以local模式跑通代码，全部功能正常后再考虑Standalone分布式集群，既能省去调试集群环境的时间，也能在答辩时证明你有能力部署真正的Spark集群。

Spark安装的第一步是确认版本匹配关系。毕设代码如果是用Spark 2.x的API写的，直接上手3.x可能遇到部分API行为变化。解压后先做三项环境配置：

export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64 export SPARK_HOME=/opt/spark-3.2.4-bin-hadoop3.2 export PATH=$SPARK_HOME/bin:$PATH

然后启动spark-shell做一个最简单的并行计算验证：

spark-shell --master local[2] sc.parallelize(1 to 100, 4).map(_ * 2).sum()

local[2]表示使用本地2个CPU核心。8G内存机器建议local[2]或local[4]，不要贪多，核心数给多了反而会因为内存不足OOM。

如果要展示Spark集群，Standalone模式是最轻量的选择，不需要HDFS也能跑。主节点和工作节点的启动命令如下：

# 主节点：-h指定Master地址，-p指定通信端口 $SPARK_HOME/sbin/start-master.sh -h 192.168.1.10 -p 7077 # 工作节点：-c指定CPU核数，-m指定可用内存 $SPARK_HOME/sbin/start-worker.sh spark://192.168.1.10:7077 -c 2 -m 2g

7077是Master与worker通信的端口，8080是Web UI端口，前者用于提交作业，后者用于浏览器查看集群状态。worker的-m 2g表示最大可分配内存，不要超过机器物理内存减去系统预留内存的值，否则系统OOM后整个节点都会掉线。

提交作业时用spark-submit，指定Master地址和主类：

spark-submit \ --master spark://192.168.1.10:7077 \ --executor-memory 1g \ --total-executor-cores 4 \ --class com.metro.analysis.FlowStat \ metro-analysis.jar \ file:///data/szmc.net-metro.csv

executor-memory 1g是合理区间。集群经验里常有一个误区：内存越大越好。实际上executor堆内存太大，JVM GC停顿时间会显著拉长，计算反而变慢。毕设数据量下，1g到2g的executor内存足够，如果发现数据量超过2g就多开executor而不是单开大内存。

大数据集群部署策略里还涉及一个细节：spark-submit的输入路径。数据放本地就用file://前缀，放HDFS则用hdfs://前缀。如果两种路径混用，最典型的问题是读取一个不存在的hdfs目录，作业启动后卡在等待状态或直接抛FileNotFoundException。用hdfs dfs -ls先确认路径存在是基本操作。

3.2 Spark中读取JSON：嵌套结构与explode展开

“spark中读取json”这个搜索点，很多人是在把API返回的JSON变成DataFrame时被卡住。JSON与CSV表格结构不同，可能出现嵌套数组。Spark需要用explode把嵌套展开后才能做分析。

假设HTTP接口返回的数据格式是：

{"station":"科学馆","records":[{"hour":8,"flow":320},{"hour":9,"flow":450}]}

读取代码：

import org.apache.spark.sql.SparkSession import org.apache.spark.sql.functions._ val spark = SparkSession.builder() .appName("MetroJsonReader") .master("local[2]") .getOrCreate() val rawDf = spark.read .option("multiLine", true) .json("file:///data/metro_records.json") val flowDf = rawDf .withColumn("record", explode(col("records"))) .select( col("station"), col("record.hour").as("hour"), col("record.flow").as("flow") ) flowDf.show(10)

explode在这里把一行中的records数组展开为多行，每个数组元素生成一行。col("record.hour")用于访问STRUCT子字段。multiLine选项处理JSON跨行的情况，一条JSON对象占多行时必须开启，否则Spark把每个物理行都当成独立JSON对象解析，结果就是大量解析失败。

如果文件是每行一个JSON对象而非一个大数组，就不要用multiLine，直接option("multiLine", false)。两种方式的取舍是：文件整体是一个大JSON用true；文件每行一个JSON对象用false。

3.3 Spark读取CSV：显式Schema与类型推断的取舍

CSV读取是毕设中最常见的入口，代码很简单，但坑都在参数上。先看显式schema的写法：

import org.apache.spark.sql.SparkSession import org.apache.spark.sql.types._ val spark = SparkSession.builder() .appName("MetroFlowStat") .master("local[4]") .config("spark.sql.shuffle.partitions", "8") .getOrCreate() val schema = StructType(Array( StructField("station_code", StringType, true), StructField("station_name", StringType, true), StructField("line_no", IntegerType, true), StructField("direction", StringType, true), StructField("ticket_type", StringType, true), StructField("in_time", TimestampType, true), StructField("out_time", TimestampType, true), StructField("flow_count", IntegerType, true) )) val df = spark.read .option("header", true) .option("dateFormat", "yyyy-MM-dd HH:mm:ss") .schema(schema) .csv("file:///data/szmc.net-metro.csv")

StructField的第三个参数nullable=true表示该字段允许null。in_time用TimestampType，如果CSV里的时间格式与dateFormat不匹配，Spark解析为null而不是报错，这是最容易“看起来成功实则失败”的地方。排查方式是用.filter(col("in_time").isNull).count()统计null数量，如果数字很大就要检查数据格式。

与显式schema相对的是inferSchema=true自动推断类型。它省事，但Spark需要额外扫描一遍数据来推断类型，数据量大时浪费时间。毕设代码里更推荐显式定义schema，一来速度快，二来类型错误在读取阶段就暴露出来，不用等到聚合时才发现某列是字符串，排查成本更低。

3.4 客流高峰统计实战：从DataFrame到聚合结果

现在拿真实CSV做一个完整的客流高峰统计：计算每个站点在早高峰(7-9点)和晚高峰(17-19点)的总客流量，输出Top20。

import org.apache.spark.sql.functions._ val peakStats = df .withColumn("hour", hour(col("in_time"))) .withColumn( "period", when(col("hour").between(7, 9), "morning_peak") .when(col("hour").between(17, 19), "evening_peak") .otherwise("off_peak") ) .filter(col("period") =!= "off_peak") .groupBy("station_name", "period") .agg( sum("flow_count").as("total_flow"), avg("flow_count").as("avg_flow") ) .orderBy(col("total_flow").desc) peakStats.show(20, false)

计算逻辑说明如下。hour函数从Timestamp列提取小时整数，between包含边界值，也就是说7到9包含7点和9点。when分支相当于CASE WHEN，用otherwise兜底非高峰时段。groupBy按站点和时段两个维度分组，sum和avg分别算出总量与均值，orderBy按总量降序排列取Top20。

spark.sql.shuffle.partitions这个参数在刚才的session配置里被设置成8，用于控制shuffle时数据分成多少个分区。系统默认200是给大数据准备的，这份数据量只有几十万行，200个task大部分是空的，白白占用调度开销。本地或者小数据场景把它降到8到16能明显提速。但如果之后要切换到完整大数据场景，记得把这个值改回100以上，否则并行度可能成为新的瓶颈。

执行后如果结果为空，有一个常见排查顺序。第一步df.count()确认CSV读取了多少行。第二步df.select("in_time").filter(col("in_time").isNotNull).limit(5).show()确认时间字段解析成功。第三步把period直接取出打印，确认when分支的过滤条件写对没有。这三步配合基本可以定位绝大多数空结果问题。

3.5 MLlib客流预测：构建训练集的最小实现

预测是毕设里的一个闪光点，用MLlib做线性回归即可，不要再引入深度学习框架。

预测思路是利用历史时间序列，用前四个小时客流量预测第五个小时客流量，构造训练数据时使用lag窗口函数：

import org.apache.spark.sql.expressions.Window import org.apache.spark.ml.feature.VectorAssembler import org.apache.spark.ml.regression.LinearRegression val tsDf = df .groupBy("station_code", "hour") .agg(sum("flow_count").as("flow")) .orderBy("station_code", "hour") val featureDf = tsDf .withColumn("lag1", lag("flow", 1).over(Window.partitionBy("station_code").orderBy("hour"))) .withColumn("lag2", lag("flow", 2).over(Window.partitionBy("station_code").orderBy("hour"))) .withColumn("lag3", lag("flow", 3).over(Window.partitionBy("station_code").orderBy("hour"))) .withColumn("lag4", lag("flow", 4).over(Window.partitionBy("station_code").orderBy("hour"))) .filter(col("lag4").isNotNull) val assembler = new VectorAssembler() .setInputCols(Array("lag1", "lag2", "lag3", "lag4")) .setOutputCol("features") val finalDf = assembler.transform(featureDf) val Array(train, test) = finalDf.randomSplit(Array(0.8, 0.2), seed = 42) val lr = new LinearRegression() .setFeaturesCol("features") .setLabelCol("flow") .setMaxIter(50) .setRegParam(0.01) val model = lr.fit(train) model.transform(test).select("flow", "prediction").show(10)

这段代码有几个关键点值得说明。

lag函数必须配合Window使用。partitionBy("station_code")意味着每个站点单独计算滞后特征，orderBy("hour")确保按时间顺序取前面的值。如果漏掉partitionBy，所有站点的数据会被打乱来计算lag，预测完全失真。

滞后特征生成后前4行必然是null，因为历史数据不够。filter(lag4.isNotNull)把不完整记录剔除，否则训练时模型遇到null特征会直接报错或丢弃数据。

randomSplit切分训练测试集时seed=42是故意的，为了让每次运行的划分结果一致。如果seed不固定，每次运行评估指标不同，答辩时无法复现报告中的数字。

LinearRegression的RegParam是L2正则化系数。0.01是一个保守值，能在客流这种有明显周期性的数据上起到防止过拟合的作用。如果预测结果波动很大，可以把RegParam提升到0.1或0.5，或者改用GBTRegressor梯度提升树。GBT不用做特征标准化，对非线性客流曲线通常表现更好。

4. 避坑指南：五个常见翻车现场与排查路径

4.1 现象：Spark作业日志没报错，但结果全null

现象：spark-submit之后日志没有error，但输出结果里客流相关字段全是null。

原因：这个地方容易误导人，日志是绿色的，看着像成功。时间字段解析失败是首要嫌疑。dateFormat与CSV实际格式不一致时，TimestampType解析会返回null而不是抛异常。另外，CSV文件如果带UTF-8 BOM，表头第一列字段名会带不可见字符，导致该列匹配失败，所有行这一字段都查不到。

排查路径要按顺序走。先df.printSchema()看时间列类型；再df.select("in_time").filter(col("in_time").isNotNull).count()看有效值有多少，如果为0就是解析问题；最后用head -3 szmc.net-metro.csv看原始时间长什么样。BOM处理可以用sed -i 's/^\xEF\xBB\xBF//'去掉BOM。这个小步骤命令行几秒完成，但每年有大量人卡在它上面。

还有一个容易被忽略的坑是CSV里同一个字段的时间格式混用，比如一半是2024-01-01 12:00:00，一半是2024/01/01 12:00:00。Spark内部用SimpleDateFormat解析，遇到第二种格式就无法识别。处理方式是用sed做一次文本替换统一格式再重新读取。如果混用属于业务数据特性，就要在代码里用when加regexp_replace做额外处理，但我建议能在数据清洗时就解决，不要拖到Spark代码里增加复杂度。

4.2 现象：HBase写入慢到无法接受，每秒只有几百条

现象：用代码批量put一百万条记录，速度只有每秒几百条，作业跑数小时。

原因：第一个是建表没有预分区。HBase默认单region，所有写入热点在同一RegionServer，即使集群有5个节点也只有一台在干活。第二个是使用逐条put提交，每条put都发起一次网络RPC，RTT开销完全掩盖了写入吞吐量。

解决分两步。第一步使用SPLITS预分区重建表：

disable 'metro_flow' drop 'metro_flow' create 'metro_flow', {NAME => 'info'}, {SPLITS => ['01', '02', '03', '04', '05']}

分区数不必贪多，按RowKey前缀的实际分布来定。分区边界设在不常出现的值上反而会打散写入热点，导致各region数据量不均衡。第二步用BufferedMutator批量写，把缓冲区攒满再统一提交：

BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf("metro_flow")); params.writeBufferSize(1024 * 1024 * 4); BufferedMutator mutator = conn.getBufferedMutator(params); Put put = new Put(Bytes.toBytes(rowkey)); put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("flow"), Bytes.toBytes("120")); mutator.mutate(put); mutator.flush();

writeBufferSize设为4MB表示缓冲区攒够4MB批量发送一次，相比逐条put性能提升一个数量级。flush时机在循环结束后再做，不要每条都flush，否则又退化成逐条提交。

4.3 现象：HBase shell能查到数据，Spark读出来却是空表

现象：HBase命令行执行scan有记录，但Spark作业读取同一个表返回的DataFrame行数为0。

原因：连接器版本与服务端不兼容是首要怀疑对象。HBase 2.x装旧版1.x的连接器，Spark执行时出现NoSuchMethodError或者加载不到TableInputFormat。另一个可能性是列族名称不匹配，HBase表定义用的是info，代码Spark读的时候用的是cf，没有实际指定正确列。

解决时先做版本核对。打开项目的依赖配置，看hbase-client、hbase-server、spark-hbase的版本坐标与目标集群是否匹配。然后写一个最小读取代码，不进行任何过滤，直接读10行打印结果，确认能正常拉数据后，再扩大查询范围。还有一个小坑是连接HBase需要的zookeeper.quorum配置，代码里没有指定ZooKeeper地址时连接会一直超时，表现为“Spark读不到”，实际上连HBase都没连上。

4.4 现象：窗口函数本地能跑，上集群却OOM

现象：本地小数据量测试时lag窗口正常，到集群处理完整数据后，作业运行时间暴涨甚至OOM。

原因：窗口函数的shuffle阶段会把partitionBy键相同的所有数据拉到一个executor上计算。如果某个站的客流历史数据特别多，比如中心枢纽站，数据倾斜就出现了。Web UI上看executor的Shuffle Read Bytes，某个任务的数据量远大于其他任务，就是倾斜的直接证据。

解决方法是让partitionBy的键更具体。原先是partitionBy("station_code")，改成partitionBy("station_code", "line_no")，使每份分区数据量更均衡。再配合spark.sql.shuffle.partitions参数调整shuffle分区数。如果倾斜仍然严重，可以对数据做repartition按站点hash重排，但这是最后手段，会增加额外一轮shuffle。毕设场景下数据量不大，通常不需要走到这步，但答辩中展示出你知道倾斜怎么排查，是一个明显的加分项。

4.5 现象：spark-submit找不到主类或加载不到依赖

现象：项目在IDEA里运行顺利，打包成jar后提交Spark集群报ClassNotFoundException或“Failed to find main class”。

原因：IDEA运行时自动打包了依赖classpath，但spark-submit不会。要么是主类本身没有被打进jar，要么是第三方库没有打包成fat jar。还有一种情况是assembly打包时多个jar的META-INF/services冲突，导致运行时服务加载器读不到实现类。

解决：使用maven-shade-plugin或sbt-assembly插件构建fat jar，确保jar内包含全部第三方依赖。主类在插件配置中要写完全限定名。然后先本地验证：

spark-submit --master local[2] --class com.metro.analysis.FlowStat metro-analysis.jar file:///data/szmc.net-metro.csv

本地通过后再换master地址，能快速验证打包正确性。如果本地都起不来，就没必要上集群排查了。需要特别注意排除Spark自身的jar，因为Spark运行环境已经提供它们，重复打入会造成class冲突。

5. 数据对拍与答辩验证：从能跑到能讲的三件事

5.1 用awk对拍验证聚合结果

Spark跑完后，需要对结果做一次独立验证。我用得最多的方法是对拍：把同一份CSV用awk做一个轻量汇总，与Spark输出对比。由于CSV字段顺序固定，awk脚本很短：

awk -F, 'NR>1 {sum[$2]+=$8} END {for (s in sum) print s, sum[s]}' szmc.net-metro.csv | sort -k2 -rn | head -20

这个命令按第二个字段（站点名）分组，把第八个字段（客流量）累加，排序后取前20。如果这20个站点的排名与Spark结果一致，主流程基本可靠。有差异时优先排查时间过滤和字段错位问题。Spark的DataFrame计算和awk这类独立工具的结果互相印证，是最直接的可信度证明。

5.2 HTTP接口与调试文件的使用

szt-api.http文件是HTTP接口调试文件。JetBrains系IDE可以直接点击运行，VS Code需要安装REST Client插件。演示时建议把接口请求、Spark结果、可视化三块连成一个流程：先通过HTTP接口确定数据范围，再调用Spark作业产出指标，最后把指标输出到图表。这样能同时展示项目的数据采集能力和计算能力。

接口验证有一个细节：HTTP接口返回JSON时，先用格式化工具确认字段名与Spark读取的列名一致。字段名大小写不一致是导致接口与数据链路脱节的常见原因。