基于Spark的交通智能分析系统：从数据清洗到拥堵指数计算

简介：本资源为基于Spark的交通智能分析系统毕业设计完整工程包，面向大数据、计算机相关专业学生及需要实战项目的开发者，帮助解决交通流数据处理、拥堵分析与流量预测等场景下的系统搭建问题。包内共339个文件，以163个dat数据文件、129个class编译文件、13个scala与8个java源码为主，另含xml配置、properties属性文件及少量txt说明，压缩包约1.45MB，覆盖数据采集、预处理、Spark SQL与Streaming分析、MLlib建模及可视化等模块。已有124人学习下载。读者可参考其目录结构与源码组织，理解RDD与DataFrame并行计算、实时流处理、交通热点聚类与事故异常检测的实现思路，并借鉴分区调优与性能监控经验，作为课程设计或毕业设计的可运行参考方案。

1. 从一份 .zip 说起：交通智能分析系统到底在算什么

早晚高峰的出租车 GPS 轨迹、卡口过车记录、公交刷卡流水，这些数据每天都在以亿级条数增长。单机跑 pandas 处理一个城市的网约车订单，内存直接爆掉，这是很多人第一次接触「基于 Spark 的交通智能分析系统」时的真实起点。这个标题背后要解决的核心问题很具体：把分散、脏乱、体量巨大的交通原始数据，用 Spark 做分布式清洗、聚合和指标计算，最终产出路段拥堵指数、区域热力、出行 OD 矩阵这类能支撑决策的结果。它适合两类人：一类是正在做大数据课程设计或毕业设计的学生，需要一套能跑通、能讲清楚链路的方案；另一类是刚转大数据的后端或数据分析工程师，想找一个真实场景把 Spark SQL、DataFrame、集群提交这些知识点串起来。整条链路不神秘，难的是数据清洗的边界和参数调优的取舍，下面按落地顺序拆开讲。

2. 交通数据从哪来、长什么样：先摸清数据源再谈 Spark

2.1 三类主流交通数据源的结构差异

做交通智能分析，第一步不是写代码，是搞清楚手里的数据长什么样。常见的三类数据源结构差别很大，直接决定了后面 Spark 怎么读、怎么切分。

第一类是网约车或出租车的 GPS 轨迹数据，典型字段包括车辆 ID、订单 ID、时间戳、经纬度、载客状态。这类数据的特点是行数极多、单行很小，一个中等城市一天轻松上千万条，而且时间戳格式经常不统一，有秒级也有毫秒级。

第二类是卡口过车记录，字段一般是卡口编号、车牌、过车时间、车道号。它的价值在于能算出路段流量和平均车速，但车牌脱敏、卡口坐标缺失是常态。

第三类是公交刷卡和地铁闸机数据，字段是卡号、线路、站点、交易时间。这类数据相对干净，但涉及换乘推断，逻辑复杂。

选型上，GPS 轨迹适合做 OD 分析和拥堵指数，卡口适合做路段流量，公交刷卡适合做客流分析。一个完整的交通智能分析系统通常不会只吃一种数据，而是以 GPS 为主、卡口为辅做交叉验证。

2.2 用 Spark 读取 JSON 和 CSV 的最小可跑代码

热搜里「spark中读取json」出现频率很高，说明很多人卡在第一步。下面这段代码用 PySpark 读取一份 JSON 格式的轨迹数据，并打印 schema，这是所有后续处理的基础。

from pyspark.sql import SparkSession from pyspark.sql.functions import col, to_timestamp # 创建 SparkSession，本地模式方便调试，集群上改成 yarn spark = SparkSession.builder \ .appName("TrafficAnalysis") \ .master("local[*]") \ .config("spark.sql.shuffle.partitions", "200") \ .getOrCreate() # 读取 JSON 轨迹数据，multiLine 用于处理跨行 JSON df = spark.read \ .option("multiLine", "true") \ .json("hdfs:///data/traffic/gps/*.json") # 打印 schema，确认字段类型是否符合预期 df.printSchema() # 统一时间戳格式，原始数据可能是字符串 df_clean = df.withColumn("event_time", to_timestamp(col("timestamp"), "yyyy-MM-dd HH:mm:ss")) df_clean.select("vehicle_id", "event_time", "lng", "lat").show(5)

逻辑说明：master("local[*]")表示用本机所有核心跑，方便先验证逻辑；上集群时改成yarn并去掉这行。spark.sql.shuffle.partitions默认是 200，小数据量下反而拖慢速度，调试阶段可以调到 8 或 16。multiLine参数只在 JSON 跨行时才需要，普通一行一条 JSON 不要开，开了会强制单分区读取，性能断崖式下跌，这是血泪经验。

参数说明：to_timestamp的格式串必须和原始数据完全一致，差一个字符就全变 null。如果原始时间戳是毫秒级数字，改用(col("ts")/1000).cast("timestamp")。读取 CSV 时用spark.read.option("header", "true").csv(...)，注意分隔符和编码，中文乱码一般是没指定encoding="utf-8"。

2.3 数据质量探查：三行代码定位脏数据分布

在读进来之后、正式清洗之前，先做一次质量探查，能省掉后面大量返工。

from pyspark.sql.functions import count, when # 统计关键字段的空值率 df_clean.select([ (count(when(col(c).isNull(), c)) / count("*")).alias(c + "_null_rate") for c in ["vehicle_id", "event_time", "lng", "lat"] ]).show() # 经纬度范围校验，中国范围大致 lng 73-135, lat 3-54 df_clean.filter( (col("lng") < 73) | (col("lng") > 135) | (col("lat") < 3) | (col("lat") > 54) ).count()

逻辑说明：第一段用列表推导批量算空值率，比一个个字段写快得多。第二段做范围过滤，GPS 漂移产生的异常坐标如果不剔除，后面算拥堵指数会严重失真。这一步的输出结果直接决定清洗策略：空值率超过 30% 的字段考虑丢弃，坐标越界比例超过 5% 要考虑是不是坐标系搞错了（GCJ-02 和 WGS-84 混用是常见坑）。

3. 用 Spark SQL 把原始轨迹算成拥堵指标

3.1 从原始点到路段匹配的计算链路

原始 GPS 点本身没有意义，必须映射到路段上才能算拥堵。常见做法是先把轨迹按车辆和时间排序，相邻两点连成线段，再和路网做空间匹配。完整链路是：去重去漂移 → 按车辆分组排序 → 计算相邻点距离和时间差 → 匹配路段 → 聚合到路段和时段。

这里有个关键取舍：精确的空间匹配需要 GIS 库，Spark 原生不支持，通常用网格化近似——把城市切成 500 米见方的网格，每个点落到网格里，用网格 ID 代替精确路段。精度损失可接受，性能提升一个数量级，这是我在实际项目里最常用的折中。

from pyspark.sql.window import Window from pyspark.sql.functions import lag, sqrt, pow, unix_timestamp, round as sround # 按车辆分组、按时间排序，取上一个点的经纬度和时间 w = Window.partitionBy("vehicle_id").orderBy("event_time") df_seg = df_clean.withColumn("prev_lng", lag("lng").over(w)) \ .withColumn("prev_lat", lag("lat").over(w)) \ .withColumn("prev_time", lag("event_time").over(w)) \ .filter(col("prev_lng").isNotNull()) # 近似计算两点距离（单位：米），1 度纬度约 111km df_seg = df_seg.withColumn( "distance_m", sqrt(pow((col("lng") - col("prev_lng")) * 111000 * 0.85, 2) + pow((col("lat") - col("prev_lat")) * 111000, 2)) ).withColumn( "time_diff_s", unix_timestamp("event_time") - unix_timestamp("prev_time") ) # 过滤异常速度：超过 120km/h 视为漂移 df_seg = df_seg.filter((col("distance_m") / col("time_diff_s")) < 33.3)

逻辑说明：lag窗口函数取上一个点，partitionBy保证不跨车辆。经度方向乘 0.85 是 cos(纬度) 的近似，中国大部分城市纬度在 30 度左右，cos(30°)≈0.87，取 0.85 够用。速度过滤阈值 33.3 m/s 对应 120 km/h，超过的基本是 GPS 跳点。

参数说明：窗口函数在大数据量下会触发 shuffle，partitionBy的字段基数不能太大，车辆 ID 通常几十万，可以接受。如果单辆车轨迹点超过百万，建议先按天分区再算，避免单分区数据倾斜。

3.2 路段拥堵指数的聚合与 Spark SQL 写法

有了每段轨迹的速度，就可以聚合到路段和时段。拥堵指数常用定义是「自由流速度 / 实际速度」，自由流速度取该路段历史 85 分位速度。

# 注册临时视图，用 Spark SQL 做聚合 df_seg.createOrReplaceTempView("traffic_seg") spark.sql(""" SELECT grid_id, date_format(event_time, 'yyyy-MM-dd HH') AS hour_bucket, count(*) AS sample_cnt, avg(distance_m / time_diff_s) AS avg_speed, percentile_approx(distance_m / time_diff_s, 0.85) AS free_flow_speed FROM traffic_seg WHERE time_diff_s > 0 AND time_diff_s < 300 GROUP BY grid_id, date_format(event_time, 'yyyy-MM-dd HH') """).createOrReplaceTempView("hourly_speed") # 计算拥堵指数 result = spark.sql(""" SELECT grid_id, hour_bucket, sample_cnt, avg_speed, free_flow_speed, round(free_flow_speed / avg_speed, 2) AS congestion_index FROM hourly_speed WHERE sample_cnt > 10 ORDER BY congestion_index DESC """) result.show(20)

逻辑说明：percentile_approx比精确分位数快得多，误差在 1% 以内，交通场景完全够用。sample_cnt > 10是过滤样本太少的网格，否则一个点算出来的指数没有统计意义。time_diff_s < 300是剔除长时间停留（比如停车熄火）导致的异常。

参数说明：percentile_approx的第二个参数是分位点，0.85 是交通领域算自由流的惯例。date_format的粒度决定聚合窗口，按小时适合看趋势，按 15 分钟适合看实时拥堵。如果数据量特别大，建议先按天聚合再按小时，减少 shuffle 数据量。

3.3 结果落库与可视化前的数据整形

算完的拥堵指数要落到 Hive 或 MySQL 供前端查询。落 Hive 用write.saveAsTable，落 MySQL 用 JDBC。

# 写入 Hive 分区表，按日期分区方便增量更新 result.write \ .mode("overwrite") \ .partitionBy("hour_bucket") \ .saveAsTable("traffic.congestion_index") # 或者写入 MySQL 供 BI 工具读取 result.write \ .format("jdbc") \ .option("url", "jdbc:mysql://localhost:3306/traffic") \ .option("dbtable", "congestion_index") \ .option("user", "root") \ .option("password", "******") \ .mode("append") \ .save()

逻辑说明：partitionBy按小时分区，查询时能分区裁剪。写 MySQL 时mode("append")是增量追加，全量覆盖用overwrite。JDBC 写入的并行度由numPartitions控制，不设置的话默认单连接写入，几百万条能写到天荒地老，这是另一个常见翻车点。

参数说明：JDBC 写入建议加.option("batchsize", "5000")和.option("numPartitions", "10")，前者控制每批提交条数，后者控制并发连接数。注意 MySQL 连接数上限，numPartitions 不要超过数据库 max_connections 的 70%。

4. 集群搭建与提交：从 local 到 yarn 的避坑清单

4.1 Spark 集群搭建的最小配置与内存参数

热搜里「spark集群搭建」和「spark内存」是高频词，说明这是卡人的地方。一个能跑交通分析的最小集群，通常 3 台机器：1 台 Master，2 台 Worker，每台 16 核 64G 起步。交通数据 shuffle 量大，内存给不够会频繁 spill 到磁盘，速度直接掉一半。

核心参数就几个：spark.executor.memory控制每个 executor 的堆内存，spark.executor.cores控制每个 executor 的核数，spark.executor.instances控制 executor 数量。经验公式是每个核配 4-5G 内存，比如 4 核配 16-20G。spark.driver.memory在 collect 大结果时才需要调大，平时 2-4G 够。

# 提交任务到 yarn 集群的完整命令 spark-submit \ --master yarn \ --deploy-mode cluster \ --executor-memory 16G \ --executor-cores 4 \ --num-executors 20 \ --driver-memory 4G \ --conf spark.sql.shuffle.partitions=400 \ --conf spark.sql.adaptive.enabled=true \ --conf spark.memory.fraction=0.6 \ traffic_analysis.py

逻辑说明：deploy-mode cluster把 driver 也放到集群，适合生产；调试用client方便看日志。spark.sql.adaptive.enabled=true开启自适应执行，能自动合并小分区，Spark 3.x 强烈建议开。spark.memory.fraction=0.6是执行和存储内存占比，默认 0.6，shuffle 大的场景可以调到 0.7。

参数说明：num-executors不是越大越好，受限于集群总核数和内存。20 个 executor × 4 核 = 80 核，集群得有这么多资源，否则任务排队。shuffle.partitions一般设成总核数的 2-3 倍，400 对应 80 核左右比较合理。

4.2 数据倾斜的定位与两种解法

交通数据倾斜特别常见：热门区域的轨迹点可能是冷门区域的几百倍，导致个别 task 跑几小时，其他 task 几分钟就完了。定位方法是看 Spark UI 的 stage 页面，如果某几个 task 的 shuffle read 远大于中位数，就是倾斜。

解法一：加盐。给倾斜的 key 加随机前缀打散，聚合后再去掉前缀二次聚合。

from pyspark.sql.functions import concat, lit, floor, rand # 给 grid_id 加 0-9 的随机前缀 df_salted = df_seg.withColumn( "salted_grid", concat(col("grid_id"), lit("_"), floor(rand() * 10).cast("string")) ) # 第一次聚合 partial = df_salted.groupBy("salted_grid").agg({"distance_m": "sum", "time_diff_s": "sum"}) # 去掉前缀二次聚合 from pyspark.sql.functions import split final = partial.withColumn("grid_id", split(col("salted_grid"), "_")[0]) \ .groupBy("grid_id").agg({"sum(distance_m)": "sum", "sum(time_diff_s)": "sum"})

逻辑说明：加盐把一个大 key 拆成 10 个小 key，并行度提升 10 倍。代价是多一次聚合，但相比倾斜导致的等待，完全值得。盐的个数根据倾斜程度定，一般 10-100。

解法二：广播小表。如果倾斜来自 join，把小表广播出去避免 shuffle。

from pyspark.sql.functions import broadcast # 路网表通常只有几万行，直接广播 result = df_seg.join(broadcast(road_network), on="grid_id", how="left")

参数说明：广播阈值由spark.sql.autoBroadcastJoinThreshold控制，默认 10MB，小表超过这个值不会自动广播，可以手动调大到 50MB，但别太大，driver 内存扛不住。

4.3 交通分析任务的常见报错与排查顺序

报错排查有个固定顺序，能省很多时间。先看 Spark UI 的 stage 和 task，定位是哪个阶段慢或失败；再看 executor 日志，找 OOM 或异常堆栈；最后看 driver 日志，确认是代码问题还是资源问题。

java.lang.OutOfMemoryError: GC overhead limit exceeded一般是 executor 内存不够或数据倾斜，先加内存，再加盐。Container killed by YARN for exceeding memory limits是堆外内存超了，调大spark.executor.memoryOverhead。FetchFailedException通常是 shuffle 数据太大或网络问题，调大spark.reducer.maxSizeInFlight或减少 shuffle 数据量。

5. 避坑与常见问题：那些让我返工的细节

5.1 时间戳时区导致的指标错位

现象：拥堵指数的高峰时段出现在凌晨 3 点，明显不对。原因：原始数据是 UTC 时间，Spark 默认按本地时区解析，差了 8 小时。解决：读取时统一转成 UTC 存储，展示时再转本地时区，或者用spark.sql.session.timeZone显式指定。

5.2 小文件过多拖垮 NameNode

现象：任务跑完 HDFS 上多了几万个几 KB 的小文件，下次读取慢得离谱。原因：shuffle.partitions设太大，或者按小时分区粒度过细。解决：写入前用coalesce或repartition合并分区，分区数控制在文件块大小的整数倍，一般 128MB 一个文件。

5.3 经纬度坐标系混用

现象：轨迹点全部落在城市外面，或者路段匹配率极低。原因：GPS 原始数据是 WGS-84，路网是 GCJ-02，两套坐标系差几百米。解决：统一转成一套坐标系再匹配，转换公式网上有现成的，注意别用错方向。

5.4 percentile_approx 在大数据量下的精度陷阱

现象：自由流速度算出来偏高，拥堵指数普遍偏低。原因：percentile_approx默认精度参数是 10000，数据量超过这个数会有误差，且对长尾分布不友好。解决：把精度参数调到 100000，或者对速度先做分桶再算分位数。

5.5 JDBC 写入把 MySQL 连接打满

现象：任务报Too many connections，数据库其他服务受影响。原因：numPartitions设太大，每个分区一个连接。解决：numPartitions 控制在 10 以内，或者改用批量写入模式，加rewriteBatchedStatements=true。

6. 让拥堵指数更可信：一个验证技巧和我的习惯

算出来的拥堵指数到底准不准，不能只看数字好看。我一般会做两件事验证。第一件是拿高德或百度的实时拥堵数据做交叉对比，取同一时段同一区域，看趋势是否一致，绝对值有偏差正常，趋势错了说明链路有问题。第二件是人工抽查，挑几个熟悉的早晚高峰路段，看系统算出来的高峰时段和实际是否吻合，这个笨办法往往最能发现问题。

验证时有个技巧：把拥堵指数按网格画成热力图，和实际路网叠加。如果热力图的高值区域和主干道对不上，大概率是网格和路段的映射关系错了，而不是算法问题。这一步用 Python 的 folium 或 Kepler.gl 都能快速出图。

# 导出结果供可视化，只取 Top 100 拥堵网格 top_grids = result.orderBy(col("congestion_index").desc()).limit(100).toPandas() top_grids.to_csv("top_congestion.csv", index=False)

逻辑说明：toPandas会把数据拉到 driver，只适合小结果集，Top 100 没问题，全量拉会 OOM。导出的 CSV 直接拖进 Kepler.gl 就能看热力图。

参数说明：limit的数字根据可视化工具的性能定，一般几千条以内都流畅。如果要做全量可视化，走 GeoMesa 或 PostGIS，别用 CSV。

最后说个习惯：每次调完参数，我都会把spark-submit的完整命令和对应的运行时长记在一个文本文件里，下次遇到类似数据量直接翻记录，比重新试快得多。交通数据的分布随城市、季节变化很大，没有一套参数能打天下，靠的就是这些记录攒出来的手感。希望帮到你。

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