简介面向毕业设计与课程作业的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直接传JSONGPS数据用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(inputColsfeature_cols, outputColfeatures) data assembler.transform(trip_features) kmeans KMeans(featuresColfeatures, k5, seed42, maxIter20) model kmeans.fit(data) predicted model.transform(data)KMeans的k值怎么定最笨但最可靠的方法是Elbow曲线对2到10个k分别计算WSSSE组内平方误差画图看拐点。交通数据通常k5或6比较合理太多分群会导致每个群体用户量太小没有业务意义。聚类结果要注意一个坑特征量纲不同平均速度是几十急加速次数是个位数直接跑KMeans会完全被速度字段主导。必须先做StandardScaler标准化。这个错误非常容易犯而且结果看起来还有模有样但实际上毫无意义。4. 集群部署与性能调优从本地到YARN的落地细节4.1 开发环境与集群环境的差异开发时用spark-shell --master local[*]跑通逻辑和真正上集群完全是两码事。本地模式下SparkDriver和Executor在同一个JVM里不需要序列化Kryo注册、没有网络开销、也没有资源竞争。提交到YARN后第一波问题通常是jar包冲突、executor内存不足、Shuffle文件溢出。我在实际项目中踩过最典型的坑是本地用spark.read.parquet读小文件没问题但集群上读几万个Parquet文件时如果每个文件只有几十MBHDFS 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会在容器里启动ExecutorSpark发行包只需放在客户端。这个理解对了能避免很多无谓的安装配置。提交参数里最影响资源利用率和稳定性的三个参数是spark-submit --master yarn --deploy-mode cluster \ --driver-memory 8g \ --executor-memory 12g \ --executor-cores 4 \ --num-executors 12 \ --conf spark.dynamicAllocation.enabledfalse \ --conf spark.sql.shuffle.partitions96 \ --conf spark.shuffle.service.enabledtrueExecutor内存和核数要背靠背看。如果executor-cores是4executor-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 用RedisWebSocket推送实时结果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 sroad:${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配置以下参数typecommand command/opt/spark/bin/spark-submit \ --class com.traffic.ODJob \ --master yarn \ --deploy-mode cluster \ --queue traffic \ --conf spark.yarn.maxAppAttempts2 \ /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以内。交通数据越攒越多从一开始就按分区设计存储后面做任何验证都会快很多。本文还有配套的精品资源点击获取 
长河君 · 2026/9/15 6:11:24 · 阅读 1.2k Spark交通智能分析实战：从实时流计算到集群性能调优
简介面向毕业设计与课程作业的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直接传JSONGPS数据用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(inputColsfeature_cols, outputColfeatures) data assembler.transform(trip_features) kmeans KMeans(featuresColfeatures, k5, seed42, maxIter20) model kmeans.fit(data) predicted model.transform(data)KMeans的k值怎么定最笨但最可靠的方法是Elbow曲线对2到10个k分别计算WSSSE组内平方误差画图看拐点。交通数据通常k5或6比较合理太多分群会导致每个群体用户量太小没有业务意义。聚类结果要注意一个坑特征量纲不同平均速度是几十急加速次数是个位数直接跑KMeans会完全被速度字段主导。必须先做StandardScaler标准化。这个错误非常容易犯而且结果看起来还有模有样但实际上毫无意义。4. 集群部署与性能调优从本地到YARN的落地细节4.1 开发环境与集群环境的差异开发时用spark-shell --master local[*]跑通逻辑和真正上集群完全是两码事。本地模式下SparkDriver和Executor在同一个JVM里不需要序列化Kryo注册、没有网络开销、也没有资源竞争。提交到YARN后第一波问题通常是jar包冲突、executor内存不足、Shuffle文件溢出。我在实际项目中踩过最典型的坑是本地用spark.read.parquet读小文件没问题但集群上读几万个Parquet文件时如果每个文件只有几十MBHDFS 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会在容器里启动ExecutorSpark发行包只需放在客户端。这个理解对了能避免很多无谓的安装配置。提交参数里最影响资源利用率和稳定性的三个参数是spark-submit --master yarn --deploy-mode cluster \ --driver-memory 8g \ --executor-memory 12g \ --executor-cores 4 \ --num-executors 12 \ --conf spark.dynamicAllocation.enabledfalse \ --conf spark.sql.shuffle.partitions96 \ --conf spark.shuffle.service.enabledtrueExecutor内存和核数要背靠背看。如果executor-cores是4executor-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 用RedisWebSocket推送实时结果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 sroad:${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配置以下参数typecommand command/opt/spark/bin/spark-submit \ --class com.traffic.ODJob \ --master yarn \ --deploy-mode cluster \ --queue traffic \ --conf spark.yarn.maxAppAttempts2 \ /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以内。交通数据越攒越多从一开始就按分区设计存储后面做任何验证都会快很多。本文还有配套的精品资源点击获取