简介：面向毕业设计与课程作业的Spark交通智能分析系统项目，以Apache Spark分布式计算框架为核心，完整覆盖从交通数据采集、预处理、车流量统计到异常检测与调度决策的闭环流程，适合大数据相关专业学生、毕业设计选题者以及希望快速上手实时分析开发的Spark初学者参考，同时其中的用户行为分析思路也可迁移至电商个性化推荐等业务场景。压缩包内共339个文件，包体大小仅1.45MB，以163个dat格式的原始数据文件和129个class编译后的类文件为主体，另含13个scala源码、8个java源码以及若干xml、properties、txt等配置与说明文档，便于对照源码、配置和运行产物进行学习研究。目前已有110人学习。资源虽小，却浓缩了Spark Streaming实时接入、Spark SQL聚合分析、MLlib流量预测、异常报警触发等关键模块的工程实现，读者可通过梳理源码理解分布式交通分析系统的模块划分、数据结构设计和任务调度逻辑，对独立完成同类课题或搭建实时分析原型极具参考价值。
1. 基于Spark的交通智能分析系统到底在解决什么问题
一个城市每天产生几亿条卡口过车记录、GPS轨迹点和路况上报数据，单机数据库连一天的数据都查不完，更别说做实时拥堵预测和OD分析。交通智能分析系统的核心不是"智能"，而是先把数据吞吐和计算延迟压下来：Spark在这里承担的是统一批流计算引擎的角色，用DataFrame做离线清洗，用Structured Streaming做实时指标，用GraphX做路网计算，用MLlib做轨迹聚类。这套系统落地后，交警支队能实时看到路网拥堵态势，公交集团能根据OD客流调整排班，互联网地图厂商能拿到准实时的路况特征。本文按一个可交付的工程方案来讲，从数据接入到集群调优，再到结果验证，覆盖做这个系统最关键的几个环节。
2. 交通数据接入与预处理：用Spark处理卡口、GPS和路况流
2.1 数据源分类与采集通道设计
交通智能分析系统的数据源大体分三类：卡口过车记录、浮动车GPS轨迹、路况事件流。卡口数据是结构化最强的，包含车牌、过车时间、卡口编号、车道号、车速，一般由前端设备通过Kafka上报。GPS轨迹来自出租车和网约车，字段里有经纬度、方向角、瞬时速度、载客状态，数据密度高但噪声也大。路况事件流是交警发布的事故、管制、施工信息，量小但对实时分析影响大。
采集通道上，常见做法是统一走Kafka，因为Kafka能削峰，也能让Spark Streaming和Structured Streaming共用一套topic。卡口数据用StringSerializer直接传JSON，GPS数据用Avro压缩编码，减少带宽占用。离线分析时，再从Kafka sink到HDFS，按天分目录，比如/data/traffic/camera/2024/05/20。这里需要注意，Kafka的topic分区数要和Spark的并行度匹配，通常一个topic设置8到16个分区，太多分区会造成Spark任务调度频繁，太少又发挥不了并行度。
# 创建topic示例，3副本，12分区 kafka-topics.sh --bootstrap-server kafka1:9092 \ --create --topic camera-event \ --partitions 12 --replication-factor 3分区数不是越大越好。12个分区配合Spark executor数量来调，如果executor总数是6个，每个executor能跑2个core，那12个分区刚好让每个core分到一个分区。分区过多时，Spark Shuffle阶段会产生大量小文件，后面处理反而更慢。
2.2 用DataFrame做清洗与特征工程
数据进到Spark之后，第一步是解析和清洗。卡口数据常见的问题有：车牌号包含特殊字符、过车时间为空、卡口编号不在字典表里、车速大于200km/h。GPS数据更乱，经纬度超出城市边界、方向角不在0-360范围、速度突变但前后点距离不合理，这些都是野点。
用Spark DataFrame处理这些，可比RDD写map逻辑直观得多。首先读入JSON或Parquet构建DataFrame，然后做过滤和修正。
val raw = spark.read.parquet("/data/traffic/camera/2024/05/20") val cleaned = raw .filter($"plate_no".isNotNull && length($"plate_no") >= 6) .filter($"camera_id".isin(cameraDict: _*)) .filter($"speed" >= 0 && $"speed" <= 180) .withColumn("record_time", to_timestamp($"record_time", "yyyy-MM-dd HH:mm:ss")) .withColumn("hour", hour($"record_time")) .withColumn("is_holiday", udf(isHoliday: (String) => Boolean).apply(lit("2024-05-20")))这段代码的关键在于：filter能下推到数据源层面，如果读的是Parquet且按分区裁剪，读取数据量会大幅减少；to_timestamp可以统一时间格式，避免后续窗口计算时出现类型不一致；hour提取小时特征，后面做分时段分析直接用。is_holiday这个UDF按日常经验补上了节假日维度，因为节假日的车流规律和工作日完全不同。
特征工程里还有一个常用操作是把卡口号映射到经纬度。卡口字典表存的是camera_id, lon, lat, road_id, direction，用一个Broadcast join把坐标维表分发到每个executor，避免每次shuffle。
val dict = spark.read.parquet("/data/dict/camera_dict") val withLoc = cleaned.join(broadcast(dict), Seq("camera_id"), "left_outer")broadcast优化在交通数据场景特别有效，因为卡口字典一般也就几千条，远小于driver内存上限。如果忽略这个，每次join都会引发全量shuffle，几亿条数据跑起来要多花十几分钟。
2.3 窗口计算：实时拥堵指数怎么算
实时拥堵指数最常见的定义是：某条路在t时刻的平均速度与自由流速度的比值。自由流速度一般取凌晨3点该路的平均速度，或者道路限速值。有了GPS点和卡口车速，就能用Structured Streaming的滑动窗口来算。
import org.apache.spark.sql.streaming._ val gpsStream = spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "kafka1:9092") .option("subscribe", "gps-event") .option("startingOffsets", "latest") .load() .selectExpr("CAST(value AS STRING) as json") .select(from_json($"json", gpsSchema).as("data")) .select("data.*") val trafficIndex = gpsStream .withWatermark("event_time", "2 minutes") .groupBy( window($"event_time", "5 minutes", "1 minute"), $"road_id" ) .agg( avg($"speed").as("avg_speed"), percentile_approx($"speed", 0.85).as("v85") )这里有一个必须强调的点：withWatermark设置的2分钟延迟阈值要和实际数据延迟匹配。卡口设备经常批量上报，GPS终端断网重连，数据延迟可能达到几分钟。如果watermark设太短，迟到的数据不会进入旧窗口，导致拥堵指数偏低；设太长，窗口状态在内存里堆积，容易OOM。一般先跑一天数据看延迟分布，再定阈值。
聚合后的结果输出到Redis或Kafka，下游可视化直接轮询。还有一个小技巧：第二条路线的指标（v85）代表85%分位速度，比平均速度更能反映路况的上限，很多交通工程报告里也用这个值作为道路通行能力的参考。
3. 核心分析任务：从OD分析到路径推荐
3.1 卡口OD矩阵的Spark SQL实现
OD矩阵是交通分析的经典需求：统计从某个区域出发到另一区域的车流量。卡口数据天然能构成OD：一辆车连续通过两个卡口，上一个卡口就是起点，下一个就是终点。但直接对全量数据做自连接，性能会非常差。常见做法是先用窗口函数按车辆分组，按时间排序拿到前后卡口。
WITH ordered AS ( SELECT plate_no, camera_id, record_time, LAG(camera_id) OVER (PARTITION BY plate_no ORDER BY record_time) AS prev_camera, LEAD(camera_id) OVER (PARTITION BY plate_no ORDER BY record_time) AS next_camera FROM cleaned_camera ) SELECT prev_camera AS origin, next_camera AS dest, COUNT(*) AS cnt FROM ordered WHERE prev_camera IS NOT NULL AND next_camera IS NOT NULL GROUP BY prev_camera, next_cameraLAG和LEAD窗口函数避免了对全表做join，每个分区内排序后直接取上下行，性能比自连接快一个数量级。这里要注意，如果一天的数据量超过几十亿行，ORDER BYrecord_time在全表范围内会触发大shuffle，所以最好先按日期分区，再按小时分区。
OD结果出来后，通常还要按交通小区聚合。交通小区是预先划分的地理区域，每个卡口属于一个小区。把OD矩阵的camera_id替换成zone_id，就能得到小区间的OD流。这一步用简单的map替换即可，但要注意同一个卡口可能服务两个方向，要对方向字段做处理，否则OD流量会重复统计。
3.2 基于GraphX的路径推荐与热点识别
如果有实时事件需要绕行建议，或者想计算两个卡口之间的最短通行时间，可以用Spark GraphX构建路网图。路网的顶点是路口，边是路段，边的权重是通行时间。Spark GraphX的ShortestPaths算法会计算从每个顶点到其他顶点的最短路径。
import org.apache.spark.graphx._ val roadVertices: RDD[(VertexId, (Double, Double))] = ... val roadEdges: RDD[Edge[Double]] = ... val graph = Graph(roadVertices, roadEdges) val landmarks = Array(1L, 100L, 200L) val results = ShortestPaths.run(graph, landmarks)ShortestPaths返回每个顶点到目标点的距离，但真正的路径还原还需要逆向追踪。实际项目中更常用的是Pregel API，自己写消息传递逻辑，可以同时输出路径点和通行时间。
热点识别则是用GraphX的连通组件和度数统计。卡口过车量可以构建一个"车辆共现图"：如果一辆车在5分钟内连续经过卡口A和B，就为这两点加一条边。然后跑triangleCount，三角形密集的区域说明有大量车辆在几个卡口之间折返，往往是商圈、学校或物流园区的车流热点。这个分析对规划新公交线路很有参考价值。
3.3 车辆轨迹聚类：用MLlib做驾驶行为分群
驾驶行为分群的输入是每条轨迹的特征向量，最常见的有：平均速度、速度标准差、急加速次数、急刹车次数、夜间行驶比例、平均驾驶时长。用Spark MLlib的KMeans按这些特征聚类，能分出货运司机、通勤人群、网约车司机等群体。
from pyspark.ml.clustering import KMeans from pyspark.ml.feature import VectorAssembler feature_cols = ["avg_speed", "speed_std", "hard_accel", "hard_brake", "night_ratio", "avg_duration"] assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") data = assembler.transform(trip_features) kmeans = KMeans(featuresCol="features", k=5, seed=42, maxIter=20) model = kmeans.fit(data) predicted = model.transform(data)KMeans的k值怎么定？最笨但最可靠的方法是Elbow曲线：对2到10个k分别计算WSSSE（组内平方误差），画图看拐点。交通数据通常k=5或6比较合理，太多分群会导致每个群体用户量太小，没有业务意义。
聚类结果要注意一个坑：特征量纲不同，平均速度是几十，急加速次数是个位数，直接跑KMeans会完全被速度字段主导。必须先做StandardScaler标准化。这个错误非常容易犯，而且结果看起来还有模有样，但实际上毫无意义。
4. 集群部署与性能调优：从本地到YARN的落地细节
4.1 开发环境与集群环境的差异
开发时用spark-shell --master local[*]跑通逻辑，和真正上集群完全是两码事。本地模式下SparkDriver和Executor在同一个JVM里，不需要序列化Kryo注册、没有网络开销、也没有资源竞争。提交到YARN后，第一波问题通常是jar包冲突、executor内存不足、Shuffle文件溢出。
我在实际项目中踩过最典型的坑是：本地用spark.read.parquet读小文件没问题，但集群上读几万个Parquet文件时，如果每个文件只有几十MB，HDFS NameNode压力会非常大。解决方法是先合并小文件再跑分析。
另一个典型的坑是序列化。默认Java序列化慢且占用空间大，集群环境应该用Kryo并提前注册类。
val conf = new SparkConf() .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") .set("spark.kryo.registrationRequired", "false") .set("spark.kryoserializer.buffer.max", "512m")4.2 Spark on YARN提交参数与动态资源
Spark on YARN提交是不是只需要一个Spark客户端？答案是：只要能在客户端机器上提交YARN任务，比如能从该机器读取HDFS路径和访问ResourceManager，就不需要在每台节点上单独部署Spark。YARN的NodeManager会在容器里启动Executor，Spark发行包只需放在客户端。这个理解对了，能避免很多无谓的安装配置。
提交参数里，最影响资源利用率和稳定性的三个参数是：
spark-submit --master yarn --deploy-mode cluster \ --driver-memory 8g \ --executor-memory 12g \ --executor-cores 4 \ --num-executors 12 \ --conf spark.dynamicAllocation.enabled=false \ --conf spark.sql.shuffle.partitions=96 \ --conf spark.shuffle.service.enabled=trueExecutor内存和核数要背靠背看。如果executor-cores是4，executor-memory给12g，那每个core大约3g。Spark官方推荐每个core不超过5g，因为JVM内部还有开销。同时spark.sql.shuffle.partitions默认是200，这个值对于交通数据量来说往往太大，会导致每个任务处理时间极短但启动开销很大。一般按executor总核数乘以2或3来设置，比如12个executor共48核，shuffle分区96个刚好。
动态资源分配（Dynamic Allocation）在交通实时任务里我不建议开启。因为实时作业要求低延迟，动态资源在大批量计算时扩容速度太慢，而且缩容时会把已缓存的状态丢掉，导致窗口计算缺失。离线批处理可以开，但要配合spark.shuffle.service才能正常缩容。
4.3 Spark内存模型与常见OOM排查
很多Spark OOM其实不是Executor堆内存不够，而是元数据和序列化缓冲溢出。交通数据特征工程里，最典型的有三种OOM：
第一种是Driver OOM。当collect()操作把所有结果拉到Driver，几千个分区的结果集一次性返回，超出Driver内存。解决办法是改成foreachPartition写入外部存储，或者用take抽样评估。事故现场通常是：执行到df.collect().foreach(println)，然后控制台卡死，Driver日志报java.lang.OutOfMemoryError。
第二种是Shuffle OOM。ORDER BY或GROUP BY时，单个key的数据量过大，比如某个卡口一天有上千万条记录，reduce端拉取所有数据后内存爆掉。常见解决方法是加spark.sql.autoBroadcastJoinThreshold并把大表数据按卡口号前缀分桶，或者调整spark.reducer.maxSizeInFlight和spark.shuffle.reduceLocality。
第三种是执行器堆外内存OOM。Kryo序列化、NIO Buffer和Netty都消耗堆外内存。当堆内存足够但频繁看到Direct buffer memory异常时，需要调大spark.executor.offHeap.size或减少executor上的任务并发数。
# 常见排查命令 yarn logs --applicationId application_1716000000000_1234 \ | grep -i "outofmemory\|GC overhead\|Direct buffer"与其事后排查，不如在代码里主动控制：任何mapPartitions内不要new大对象，能复用的变量提到循环外层；广播变量不超过1GB，否则Driver端序列化时间太长；repartition和coalesce要分清，前者是shuffle重新分区，后者只合并分区数据，不产生shuffle。很多看起来是OOM的问题，其实是反正则写了过宽的分区导致空任务堆积。
5. 结果可视化与调度验证：让分析结果真正进入业务
5.1 用Redis+WebSocket推送实时结果
Structured Streaming算出的拥堵指数如果只是写在Parquet里，业务方感知不到价值。常见做法是把最新指标写入Redis的有序集合或哈希，然后由WebSocket服务订阅推送。Redis里每个key代表一条路，score存时间戳，value存JSON，例如：
streamResult.foreachBatch { (batchDF, batchId) => batchDF.foreachPartition { rows => val jedis = new Jedis(redisHost, redisPort) rows.foreach { row => val key = s"road:${row.getAs[Long]("road_id")}" val value = Map( "idx" -> row.getAs[Double]("traffic_index"), "speed" -> row.getAs[Double]("avg_speed"), "ts" -> row.getAs[java.sql.Timestamp]("window_end").getTime ) jedis.zadd(key, row.getAs[java.sql.Timestamp]("window_end").getTime, value) } jedis.close() } }重点在于foreachPartition内创建连接，而不是每条数据创建一个Jedis，那样会把Redis连接池打爆。对WebSocket推送，前端直接订阅road:1001即可，不必在Spark里实现推送协议，避免强耦合。
5.2 用Azkaban调度离线分析任务
离线OD矩阵、轨迹聚类的任务通常是按天执行。直接把Spark提交命令写进Azkaban的command job很简单，但更规范的做法是拆成两步：先运行数据检查job，再运行主任务。如果前一天的数据没有到位，检查job直接失败，不浪费计算资源。
一个典型的Azkaban flow里，主job配置以下参数：
type=command command=/opt/spark/bin/spark-submit \ --class com.traffic.ODJob \ --master yarn \ --deploy-mode cluster \ --queue traffic \ --conf spark.yarn.maxAppAttempts=2 \ /data/app/traffic-analyzer-1.0.jar \ --date=${dt}date参数通过Azkaban的调度变量传入，任务重跑时只需要换一个日期值。这里的spark.yarn.maxAppAttempts要谨慎设置，如果应用因代码bug失败，重试只是反复失败，不会产出数据。一般设2，不要设更高。
5.3 验证分析结果的三个常用方法
Spark算出来的数据如果没人验证，业务方大概率不敢直接用。我一般会在交付前做这三件事：
第一，用口径对账。把Spark统计的OD总量和卡口系统导出的原始过车总量做比对，误差超过5%就说明清洗逻辑或去重逻辑有问题。常见偏差源是车牌号格式不一致导致一辆车被算成两辆，或者卡口方向字段取反而使OD方向颠倒。
第二，用时间对比。因为交通数据有明显的潮汐性，分析结果要和前一周同一天、同一时段对比。如果某条路的拥堵指数突然从1.5跳到5，可能是数据源断流或清洗规则误伤，不一定是真堵了。这个规则可以用一个简单的Spark任务来监控：
SELECT road_id, avg(traffic_index) AS today, lag(avg(traffic_index), 168) OVER (ORDER BY road_id) AS last_week FROM traffic_index_hourly WHERE dt = '2024-05-20' GROUP BY road_id HAVING abs(today - last_week) > 2.5第三，用人工抽检。随机抽样3到5辆车，把它们的轨迹从Spark分析结果中还原出来，和原始GPS记录对比。这一步最土但最有效。尤其是路径推荐结果，要验证推荐路径在真实路网中确实可达，因为路网图数据可能缺失某条新建道路，导致GraphX计算出错误路径。
最后说一个关于结果输出的细节：离线分析结果写Parquet时，要按dt/hh分区，并且用optimize writer开启列压缩，这样下游跑SQL查某一天某小时的数据，扫描量可以控制到几十MB以内。交通数据越攒越多，从一开始就按分区设计存储，后面做任何验证都会快很多。
本文还有配套的精品资源，点击获取
