简介这是一份面向高校学生与大数据初学者的Spark实战项目资料以交通智能分析系统为背景适合用作毕业设计、课程作业或大数据入门练手。项目围绕数据采集、预处理、流量分析、异常检测与决策支持五个模块展开借助Spark Streaming、Spark SQL与MLlib完成实时车流统计、拥堵预警和流量预测其分析思路也可迁移到电商用户行为挖掘等场景。压缩包共339个文件约1.45MB包含163个dat数据文件、129个class编译文件以及13个scala与8个java源码另有xml配置、properties参数文件和少量txt说明覆盖从源码到运行数据的完整结构。目前已有111人学习下载。通过研读源码与数据文件读者可掌握Spark分布式计算、实时流处理与机器学习建模的落地方法理解交通领域大数据系统的设计脉络并积累可复用的项目经验与排错思路。1. 从一份交通卡口流水说起Spark 智能分析系统到底在算什么早高峰的卡口流水一天能到千万行字段无非是过车时间、设备编号、车牌哈希、车道号、车型。单机 pandas 读到三百万行就开始喘groupby 一跑内存直接爆掉这是很多人第一次意识到需要 Spark 的时刻。基于 Spark 的交通智能分析系统本质就是把这类流水做成可查询、可聚合、可预警的指标层拥堵指数、路段平均车速、早晚高峰流量对比、异常过车识别。它适合两类人一类是要交课程设计或工程实践、需要一套能跑起来的完整链路另一类是已经有一批卡口或 GPS 数据、想用 Spark 把离线统计和准实时聚合搭出来。标题里的「设计与实现」不是写文档而是把数据从原始 CSV 一路推到能出图、能出报表的结果表。下面按我实际搭过的顺序拆开讲先讲清楚数据怎么进来、怎么分层再落到 Spark SQL 和调优参数最后把踩过的坑摆出来。2. 数据分层与表结构交通流水的 ODS、DWD、DWS 怎么切2.1 为什么不能一张宽表跑到底交通数据的天然特征是「原始流水极宽、指标极窄」。原始过车记录有几十个字段但真正用于分析的往往只有时间、路段、方向、车型四五个维度。如果所有计算都直接怼在原始表上每次跑指标都要全量扫一遍磁盘 IO 和 shuffle 都吃不消。常见做法是分三层ODS 层保留原始接入数据字段和来源一致只做格式统一DWD 层做清洗和标准化把时间戳归一到秒、把设备编号映射到路段、把无效车牌和重复过车剔掉DWS 层按「路段 时间窗」预聚合直接产出流量、平均车速、饱和度这类指标。这样上层做报表或预警时只扫 DWS数据量能降一到两个数量级。分层还有一个隐性好处排错时能定位到具体环节。指标不对先看 DWD 的清洗规则再看 DWS 的聚合口径不用在一张大宽表里猜是哪一步出的问题。我一般会把每层的分区字段统一成dt日期加hour小时这样按天回溯和按小时补数都方便。2.2 建表语句与分区设计下面这套 DDL 是我在 Hive 外表上常用的结构ODS 用文本或 Parquet 都行DWD 和 DWS 统一 Parquet 加 snappy 压缩。-- ODS原始过车流水按天分区 CREATE TABLE ods_traffic_pass ( pass_time STRING COMMENT 过车时间 yyyy-MM-dd HH:mm:ss, device_id STRING COMMENT 卡口设备编号, plate_hash STRING COMMENT 车牌哈希脱敏后, lane_no INT COMMENT 车道号, vehicle_type STRING COMMENT 车型, speed DOUBLE COMMENT 瞬时车速可能为空 ) PARTITIONED BY (dt STRING) STORED AS PARQUET; -- DWD清洗后标准流水补上路段维度 CREATE TABLE dwd_traffic_pass ( pass_ts BIGINT COMMENT 过车时间戳秒, road_id STRING COMMENT 路段编号, direction STRING COMMENT 方向上行/下行, plate_hash STRING, lane_no INT, vehicle_type STRING, speed DOUBLE ) PARTITIONED BY (dt STRING, hour STRING) STORED AS PARQUET; -- DWS路段小时级指标 CREATE TABLE dws_road_hour_metric ( road_id STRING, direction STRING, hour STRING, pass_cnt BIGINT COMMENT 过车数, avg_speed DOUBLE COMMENT 平均车速, congestion DOUBLE COMMENT 拥堵指数 0-10 ) PARTITIONED BY (dt STRING) STORED AS PARQUET;分区字段的选择直接决定后续查询能不能裁剪。dt放最外层是为了按天批量重跑hour放内层是为了小时级补数。注意 DWD 的pass_ts用 BIGINT 而不是字符串后面做时间窗聚合时省掉一次unix_timestamp转换这个细节在千万级数据上能省不少 CPU。2.3 设备到路段的映射表怎么维护卡口设备编号和路段的对应关系不是一成不变的新建设备、临时改道都会让映射失效。我一般单独建一张维表dim_device_road字段是device_id、road_id、direction、start_date、end_date用拉链方式保留历史。DWD 清洗时按pass_time落在start_date和end_date之间去 join避免用最新映射去套历史数据导致路段归属错乱。这张表数据量小可以广播到每个 executorjoin 时不会有 shuffle。3. 用 Spark SQL 把原始流水跑成拥堵指标3.1 从 ODS 到 DWD 的清洗逻辑清洗这一步的核心是去重、补维、算时间戳。去重不能简单按车牌去同一辆车短时间内多次过同一设备可能是重复上报也可能是真的绕了一圈我一般按「设备 车牌 分钟」做窗口去重保留最早一条。from pyspark.sql import SparkSession from pyspark.sql import functions as F from pyspark.sql.window import Window spark SparkSession.builder \ .appName(traffic_dwd_clean) \ .config(spark.sql.shuffle.partitions, 200) \ .config(spark.sql.adaptive.enabled, true) \ .enableHiveSupport() \ .getOrCreate() # 读 ODS 当天分区 ods spark.table(ods_traffic_pass).filter(F.col(dt) 2024-06-01) # 时间戳归一 分钟窗口去重 ods ods.withColumn(pass_ts, F.unix_timestamp(pass_time, yyyy-MM-dd HH:mm:ss)) w Window.partitionBy(device_id, plate_hash, F.floor(F.col(pass_ts) / 60)).orderBy(pass_ts) dwd ods.withColumn(rn, F.row_number().over(w)) \ .filter(F.col(rn) 1) \ .drop(rn) # 广播维表补路段 dim spark.table(dim_device_road) \ .filter(F.col(end_date).isNull() | (F.col(end_date) 2024-06-01)) dwd dwd.join(F.broadcast(dim), ondevice_id, howleft) \ .withColumn(hour, F.lpad(F.hour(pass_time), 2, 0)) \ .select(pass_ts, road_id, direction, plate_hash, lane_no, vehicle_type, speed, dt, hour) dwd.write.mode(overwrite).insertInto(dwd_traffic_pass)spark.sql.shuffle.partitions设 200 是经验值数据量在几千万行时比较稳太小会 OOM太大会产生大量小文件。spark.sql.adaptive.enabled打开后 Spark 会自动合并小分区对倾斜的卡口数据特别有用——某些主干道设备的数据量可能是支路的几十倍自适应能缓解长尾。广播维表用F.broadcast显式提示避免优化器判断失误走 shuffle join。3.2 小时级指标聚合与拥堵指数计算DWS 层的聚合是整套系统的核心。过车数直接 count平均车速用avg(speed)但要注意空值处理——很多卡口不返回瞬时车速直接 avg 会把空值算进去导致结果偏低我一般先过滤speed is not null再算同时记录有效样本数。拥堵指数没有统一公式常见做法是用「实际行程时间 / 自由流行程时间」的比值再映射到 0-10。没有行程时间数据时可以用平均车速反推设自由流车速为v_free比如 60 km/h拥堵指数 min(10, v_free / avg_speed)。这个公式粗糙但可解释适合课程设计或初期版本。dwd spark.table(dwd_traffic_pass).filter(F.col(dt) 2024-06-01) dws dwd.groupBy(road_id, direction, hour, dt).agg( F.count(*).alias(pass_cnt), F.avg(F.when(F.col(speed).isNotNull(), F.col(speed))).alias(avg_speed), F.sum(F.when(F.col(speed).isNotNull(), 1).otherwise(0)).alias(speed_samples) ) # 拥堵指数自由流 60无有效车速时置空 dws dws.withColumn( congestion, F.when(F.col(avg_speed) 0, F.least(F.lit(10.0), F.lit(60.0) / F.col(avg_speed))) .otherwise(None) ) dws.write.mode(overwrite).insertInto(dws_road_hour_metric)F.least用来封顶避免低速时指数飙到几十。speed_samples这个字段很多人会漏它的作用是让下游知道这条指标可不可信——样本数只有个位数时平均车速波动极大报表里应该标注出来。3.3 早晚高峰对比怎么查有了 DWS早晚高峰对比就是一句 SQL 的事。下面查每个路段早高峰7-9 点和晚高峰17-19 点的流量与平均车速对比。SELECT road_id, direction, SUM(CASE WHEN hour BETWEEN 07 AND 09 THEN pass_cnt ELSE 0 END) AS am_cnt, SUM(CASE WHEN hour BETWEEN 17 AND 19 THEN pass_cnt ELSE 0 END) AS pm_cnt, AVG(CASE WHEN hour BETWEEN 07 AND 09 THEN avg_speed END) AS am_speed, AVG(CASE WHEN hour BETWEEN 17 AND 19 THEN avg_speed END) AS pm_speed FROM dws_road_hour_metric WHERE dt 2024-06-01 GROUP BY road_id, direction ORDER BY am_cnt DESC LIMIT 50;这里用CASE WHEN而不是两次查询再 join减少一次扫描。AVG对avg_speed再平均是近似值严格来说应该用加权平均按 pass_cnt 加权如果对精度要求高把SUM(avg_speed * pass_cnt) / SUM(pass_cnt)换上去即可。4. 避坑与排查交通数据在 Spark 上最容易翻车的五件事4.1 数据倾斜某个卡口的数据量是别人的一百倍现象是任务卡在最后一个 reduce 阶段不动打开 Spark UI 看到某个 task 处理的数据量远超其他。原因是主干道卡口的过车量天然远高于支路groupBy 路段时数据全挤到少数分区。解决办法分两步先确认倾斜键用dwd.groupBy(road_id).count().orderBy(F.desc(count)).show(10)找出来再对倾斜键加随机前缀打散聚合完再去掉前缀二次聚合。如果用的是 Spark 3.x直接开 AQE 的倾斜处理spark.sql.adaptive.skewJoin.enabledtrue也能缓解大部分场景。4.2 时间戳时区错乱导致高峰时段偏移现象是早高峰统计算出来落在凌晨。原因是unix_timestamp默认按 JVM 时区解析集群时区是 UTC 时yyyy-MM-dd HH:mm:ss会被当成 UTC 时间和本地时间差 8 小时。解决方式是在 SparkSession 里显式设spark.sql.session.timeZoneAsia/Shanghai或者清洗时统一用to_timestamp并指定格式不要依赖默认行为。这个坑很隐蔽因为数据本身没错错的是解释方式。4.3 小文件过多拖垮 NameNode现象是跑完一天数据后 HDFS 上多出几万个几十 KB 的文件下次查询光列目录就要好几秒。原因是分区粒度太细dt hour 设备加上并行度高。解决办法是写入前用repartition控制文件数或者在 DWS 层按 dt 聚合后只写一个分区。我一般会在写入后跑一个合并脚本把小于 128MB 的文件合并掉。注意coalesce和repartition的区别前者不 shuffle 但可能造成分区不均后者 shuffle 但分布均匀按数据量选。4.4 车牌哈希后仍然能反推现象是脱敏做了但还能通过哈希碰撞或字典攻击还原。原因是用了无盐的 MD5 或 SHA1车牌空间小彩虹表一查就出来。解决方式是加固定盐值再哈希或者直接用 HMAC。盐值不要写在代码里放配置中心或环境变量。这个坑在课程设计里经常被忽略但一旦数据外流就是实打实的隐私问题。4.5 内存参数照搬网上配置导致 executor 被 kill现象是任务跑一半报Container killed by YARN for exceeding memory limits。原因是spark.executor.memory设得大但spark.executor.memoryOverhead没跟上或者spark.memory.fraction默认 0.6 导致执行内存不够。我的习惯是 executor 内存设 4-8Goverhead 给 10%-15%spark.memory.fraction保持默认先跑小数据量压测再放大。另外spark.sql.shuffle.partitions和 executor 数量要匹配200 个分区配 4 个 executor 就是每个 executor 扛 50 个分区容易 OOM。5. 把离线指标接到准实时Structured Streaming 的增量聚合技巧离线跑通之后很多人会想能不能让拥堵指数分钟级更新。Structured Streaming 是 Spark 原生的流处理入口和批处理共用一套 API迁移成本低。我的做法是把 Kafka 里的过车消息按「路段 5 分钟窗口」做增量聚合输出到 DWS 的实时表离线表负责 T1 校准两张表在查询层 union。stream spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, broker:9092) \ .option(subscribe, traffic_pass) \ .load() parsed stream.selectExpr(CAST(value AS STRING)) \ .select(F.from_json(value, pass_ts LONG, road_id STRING, direction STRING, speed DOUBLE).alias(d)) \ .select(d.*) \ .withWatermark(pass_ts, 2 minutes) agg parsed.groupBy( F.col(road_id), F.col(direction), F.window(F.col(pass_ts).cast(timestamp), 5 minutes) ).agg( F.count(*).alias(pass_cnt), F.avg(speed).alias(avg_speed) ) agg.writeStream \ .outputMode(update) \ .format(parquet) \ .option(path, /warehouse/dws_road_realtime) \ .option(checkpointLocation, /checkpoint/traffic_realtime) \ .trigger(processingTime1 minute) \ .start()withWatermark设 2 分钟是容忍乱序的边界交通数据从设备上报到进 Kafka 一般延迟在秒级2 分钟足够覆盖大部分乱序。outputMode(update)只输出变化的窗口比complete省资源。trigger设 1 分钟意味着微批间隔延迟和吞吐的折中点。checkpoint 目录必须放在可靠存储上否则重启后状态丢失会重复计算。这里有个容易忽略的点流处理的聚合结果和离线结果对不上是正常的因为窗口边界和乱序处理策略不同。我的习惯是在报表层标注数据来源实时值用于监控告警离线值用于正式统计两者不混用。另外流式写入 parquet 会产生大量小文件生产环境一般换成 Delta 或 Hudi课程设计阶段用 parquet 加定时合并也能接受。调优上流处理对 executor 数量不敏感但对内存敏感因为状态要常驻。spark.sql.streaming.stateStore.providerClass默认的 HDFSBackedStateStore 在状态大时性能一般数据量上来后可以换 RocksDB。这些参数不用一开始就调先让链路跑通压测时再逐个改。最后说个我自己的习惯每次改完聚合逻辑先拿一天的历史数据用批处理跑一遍确认指标口径对了再切到流。批流用同一套 SQL能省掉大量对账时间。这套系统值不值得做取决于你手上有没有持续产生的交通数据——有数据Spark 的分层和聚合能力能让你从「导出一张 Excel」变成「随时查任意路段任意时段」没数据先把公开的卡口数据集跑通链路再考虑接真实源。希望帮到你。本文还有配套的精品资源点击获取 基于Spark的交通智能分析系统设计与实现：从卡口流水到拥堵指标
简介这是一份面向高校学生与大数据初学者的Spark实战项目资料以交通智能分析系统为背景适合用作毕业设计、课程作业或大数据入门练手。项目围绕数据采集、预处理、流量分析、异常检测与决策支持五个模块展开借助Spark Streaming、Spark SQL与MLlib完成实时车流统计、拥堵预警和流量预测其分析思路也可迁移到电商用户行为挖掘等场景。压缩包共339个文件约1.45MB包含163个dat数据文件、129个class编译文件以及13个scala与8个java源码另有xml配置、properties参数文件和少量txt说明覆盖从源码到运行数据的完整结构。目前已有111人学习下载。通过研读源码与数据文件读者可掌握Spark分布式计算、实时流处理与机器学习建模的落地方法理解交通领域大数据系统的设计脉络并积累可复用的项目经验与排错思路。1. 从一份交通卡口流水说起Spark 智能分析系统到底在算什么早高峰的卡口流水一天能到千万行字段无非是过车时间、设备编号、车牌哈希、车道号、车型。单机 pandas 读到三百万行就开始喘groupby 一跑内存直接爆掉这是很多人第一次意识到需要 Spark 的时刻。基于 Spark 的交通智能分析系统本质就是把这类流水做成可查询、可聚合、可预警的指标层拥堵指数、路段平均车速、早晚高峰流量对比、异常过车识别。它适合两类人一类是要交课程设计或工程实践、需要一套能跑起来的完整链路另一类是已经有一批卡口或 GPS 数据、想用 Spark 把离线统计和准实时聚合搭出来。标题里的「设计与实现」不是写文档而是把数据从原始 CSV 一路推到能出图、能出报表的结果表。下面按我实际搭过的顺序拆开讲先讲清楚数据怎么进来、怎么分层再落到 Spark SQL 和调优参数最后把踩过的坑摆出来。2. 数据分层与表结构交通流水的 ODS、DWD、DWS 怎么切2.1 为什么不能一张宽表跑到底交通数据的天然特征是「原始流水极宽、指标极窄」。原始过车记录有几十个字段但真正用于分析的往往只有时间、路段、方向、车型四五个维度。如果所有计算都直接怼在原始表上每次跑指标都要全量扫一遍磁盘 IO 和 shuffle 都吃不消。常见做法是分三层ODS 层保留原始接入数据字段和来源一致只做格式统一DWD 层做清洗和标准化把时间戳归一到秒、把设备编号映射到路段、把无效车牌和重复过车剔掉DWS 层按「路段 时间窗」预聚合直接产出流量、平均车速、饱和度这类指标。这样上层做报表或预警时只扫 DWS数据量能降一到两个数量级。分层还有一个隐性好处排错时能定位到具体环节。指标不对先看 DWD 的清洗规则再看 DWS 的聚合口径不用在一张大宽表里猜是哪一步出的问题。我一般会把每层的分区字段统一成dt日期加hour小时这样按天回溯和按小时补数都方便。2.2 建表语句与分区设计下面这套 DDL 是我在 Hive 外表上常用的结构ODS 用文本或 Parquet 都行DWD 和 DWS 统一 Parquet 加 snappy 压缩。-- ODS原始过车流水按天分区 CREATE TABLE ods_traffic_pass ( pass_time STRING COMMENT 过车时间 yyyy-MM-dd HH:mm:ss, device_id STRING COMMENT 卡口设备编号, plate_hash STRING COMMENT 车牌哈希脱敏后, lane_no INT COMMENT 车道号, vehicle_type STRING COMMENT 车型, speed DOUBLE COMMENT 瞬时车速可能为空 ) PARTITIONED BY (dt STRING) STORED AS PARQUET; -- DWD清洗后标准流水补上路段维度 CREATE TABLE dwd_traffic_pass ( pass_ts BIGINT COMMENT 过车时间戳秒, road_id STRING COMMENT 路段编号, direction STRING COMMENT 方向上行/下行, plate_hash STRING, lane_no INT, vehicle_type STRING, speed DOUBLE ) PARTITIONED BY (dt STRING, hour STRING) STORED AS PARQUET; -- DWS路段小时级指标 CREATE TABLE dws_road_hour_metric ( road_id STRING, direction STRING, hour STRING, pass_cnt BIGINT COMMENT 过车数, avg_speed DOUBLE COMMENT 平均车速, congestion DOUBLE COMMENT 拥堵指数 0-10 ) PARTITIONED BY (dt STRING) STORED AS PARQUET;分区字段的选择直接决定后续查询能不能裁剪。dt放最外层是为了按天批量重跑hour放内层是为了小时级补数。注意 DWD 的pass_ts用 BIGINT 而不是字符串后面做时间窗聚合时省掉一次unix_timestamp转换这个细节在千万级数据上能省不少 CPU。2.3 设备到路段的映射表怎么维护卡口设备编号和路段的对应关系不是一成不变的新建设备、临时改道都会让映射失效。我一般单独建一张维表dim_device_road字段是device_id、road_id、direction、start_date、end_date用拉链方式保留历史。DWD 清洗时按pass_time落在start_date和end_date之间去 join避免用最新映射去套历史数据导致路段归属错乱。这张表数据量小可以广播到每个 executorjoin 时不会有 shuffle。3. 用 Spark SQL 把原始流水跑成拥堵指标3.1 从 ODS 到 DWD 的清洗逻辑清洗这一步的核心是去重、补维、算时间戳。去重不能简单按车牌去同一辆车短时间内多次过同一设备可能是重复上报也可能是真的绕了一圈我一般按「设备 车牌 分钟」做窗口去重保留最早一条。from pyspark.sql import SparkSession from pyspark.sql import functions as F from pyspark.sql.window import Window spark SparkSession.builder \ .appName(traffic_dwd_clean) \ .config(spark.sql.shuffle.partitions, 200) \ .config(spark.sql.adaptive.enabled, true) \ .enableHiveSupport() \ .getOrCreate() # 读 ODS 当天分区 ods spark.table(ods_traffic_pass).filter(F.col(dt) 2024-06-01) # 时间戳归一 分钟窗口去重 ods ods.withColumn(pass_ts, F.unix_timestamp(pass_time, yyyy-MM-dd HH:mm:ss)) w Window.partitionBy(device_id, plate_hash, F.floor(F.col(pass_ts) / 60)).orderBy(pass_ts) dwd ods.withColumn(rn, F.row_number().over(w)) \ .filter(F.col(rn) 1) \ .drop(rn) # 广播维表补路段 dim spark.table(dim_device_road) \ .filter(F.col(end_date).isNull() | (F.col(end_date) 2024-06-01)) dwd dwd.join(F.broadcast(dim), ondevice_id, howleft) \ .withColumn(hour, F.lpad(F.hour(pass_time), 2, 0)) \ .select(pass_ts, road_id, direction, plate_hash, lane_no, vehicle_type, speed, dt, hour) dwd.write.mode(overwrite).insertInto(dwd_traffic_pass)spark.sql.shuffle.partitions设 200 是经验值数据量在几千万行时比较稳太小会 OOM太大会产生大量小文件。spark.sql.adaptive.enabled打开后 Spark 会自动合并小分区对倾斜的卡口数据特别有用——某些主干道设备的数据量可能是支路的几十倍自适应能缓解长尾。广播维表用F.broadcast显式提示避免优化器判断失误走 shuffle join。3.2 小时级指标聚合与拥堵指数计算DWS 层的聚合是整套系统的核心。过车数直接 count平均车速用avg(speed)但要注意空值处理——很多卡口不返回瞬时车速直接 avg 会把空值算进去导致结果偏低我一般先过滤speed is not null再算同时记录有效样本数。拥堵指数没有统一公式常见做法是用「实际行程时间 / 自由流行程时间」的比值再映射到 0-10。没有行程时间数据时可以用平均车速反推设自由流车速为v_free比如 60 km/h拥堵指数 min(10, v_free / avg_speed)。这个公式粗糙但可解释适合课程设计或初期版本。dwd spark.table(dwd_traffic_pass).filter(F.col(dt) 2024-06-01) dws dwd.groupBy(road_id, direction, hour, dt).agg( F.count(*).alias(pass_cnt), F.avg(F.when(F.col(speed).isNotNull(), F.col(speed))).alias(avg_speed), F.sum(F.when(F.col(speed).isNotNull(), 1).otherwise(0)).alias(speed_samples) ) # 拥堵指数自由流 60无有效车速时置空 dws dws.withColumn( congestion, F.when(F.col(avg_speed) 0, F.least(F.lit(10.0), F.lit(60.0) / F.col(avg_speed))) .otherwise(None) ) dws.write.mode(overwrite).insertInto(dws_road_hour_metric)F.least用来封顶避免低速时指数飙到几十。speed_samples这个字段很多人会漏它的作用是让下游知道这条指标可不可信——样本数只有个位数时平均车速波动极大报表里应该标注出来。3.3 早晚高峰对比怎么查有了 DWS早晚高峰对比就是一句 SQL 的事。下面查每个路段早高峰7-9 点和晚高峰17-19 点的流量与平均车速对比。SELECT road_id, direction, SUM(CASE WHEN hour BETWEEN 07 AND 09 THEN pass_cnt ELSE 0 END) AS am_cnt, SUM(CASE WHEN hour BETWEEN 17 AND 19 THEN pass_cnt ELSE 0 END) AS pm_cnt, AVG(CASE WHEN hour BETWEEN 07 AND 09 THEN avg_speed END) AS am_speed, AVG(CASE WHEN hour BETWEEN 17 AND 19 THEN avg_speed END) AS pm_speed FROM dws_road_hour_metric WHERE dt 2024-06-01 GROUP BY road_id, direction ORDER BY am_cnt DESC LIMIT 50;这里用CASE WHEN而不是两次查询再 join减少一次扫描。AVG对avg_speed再平均是近似值严格来说应该用加权平均按 pass_cnt 加权如果对精度要求高把SUM(avg_speed * pass_cnt) / SUM(pass_cnt)换上去即可。4. 避坑与排查交通数据在 Spark 上最容易翻车的五件事4.1 数据倾斜某个卡口的数据量是别人的一百倍现象是任务卡在最后一个 reduce 阶段不动打开 Spark UI 看到某个 task 处理的数据量远超其他。原因是主干道卡口的过车量天然远高于支路groupBy 路段时数据全挤到少数分区。解决办法分两步先确认倾斜键用dwd.groupBy(road_id).count().orderBy(F.desc(count)).show(10)找出来再对倾斜键加随机前缀打散聚合完再去掉前缀二次聚合。如果用的是 Spark 3.x直接开 AQE 的倾斜处理spark.sql.adaptive.skewJoin.enabledtrue也能缓解大部分场景。4.2 时间戳时区错乱导致高峰时段偏移现象是早高峰统计算出来落在凌晨。原因是unix_timestamp默认按 JVM 时区解析集群时区是 UTC 时yyyy-MM-dd HH:mm:ss会被当成 UTC 时间和本地时间差 8 小时。解决方式是在 SparkSession 里显式设spark.sql.session.timeZoneAsia/Shanghai或者清洗时统一用to_timestamp并指定格式不要依赖默认行为。这个坑很隐蔽因为数据本身没错错的是解释方式。4.3 小文件过多拖垮 NameNode现象是跑完一天数据后 HDFS 上多出几万个几十 KB 的文件下次查询光列目录就要好几秒。原因是分区粒度太细dt hour 设备加上并行度高。解决办法是写入前用repartition控制文件数或者在 DWS 层按 dt 聚合后只写一个分区。我一般会在写入后跑一个合并脚本把小于 128MB 的文件合并掉。注意coalesce和repartition的区别前者不 shuffle 但可能造成分区不均后者 shuffle 但分布均匀按数据量选。4.4 车牌哈希后仍然能反推现象是脱敏做了但还能通过哈希碰撞或字典攻击还原。原因是用了无盐的 MD5 或 SHA1车牌空间小彩虹表一查就出来。解决方式是加固定盐值再哈希或者直接用 HMAC。盐值不要写在代码里放配置中心或环境变量。这个坑在课程设计里经常被忽略但一旦数据外流就是实打实的隐私问题。4.5 内存参数照搬网上配置导致 executor 被 kill现象是任务跑一半报Container killed by YARN for exceeding memory limits。原因是spark.executor.memory设得大但spark.executor.memoryOverhead没跟上或者spark.memory.fraction默认 0.6 导致执行内存不够。我的习惯是 executor 内存设 4-8Goverhead 给 10%-15%spark.memory.fraction保持默认先跑小数据量压测再放大。另外spark.sql.shuffle.partitions和 executor 数量要匹配200 个分区配 4 个 executor 就是每个 executor 扛 50 个分区容易 OOM。5. 把离线指标接到准实时Structured Streaming 的增量聚合技巧离线跑通之后很多人会想能不能让拥堵指数分钟级更新。Structured Streaming 是 Spark 原生的流处理入口和批处理共用一套 API迁移成本低。我的做法是把 Kafka 里的过车消息按「路段 5 分钟窗口」做增量聚合输出到 DWS 的实时表离线表负责 T1 校准两张表在查询层 union。stream spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, broker:9092) \ .option(subscribe, traffic_pass) \ .load() parsed stream.selectExpr(CAST(value AS STRING)) \ .select(F.from_json(value, pass_ts LONG, road_id STRING, direction STRING, speed DOUBLE).alias(d)) \ .select(d.*) \ .withWatermark(pass_ts, 2 minutes) agg parsed.groupBy( F.col(road_id), F.col(direction), F.window(F.col(pass_ts).cast(timestamp), 5 minutes) ).agg( F.count(*).alias(pass_cnt), F.avg(speed).alias(avg_speed) ) agg.writeStream \ .outputMode(update) \ .format(parquet) \ .option(path, /warehouse/dws_road_realtime) \ .option(checkpointLocation, /checkpoint/traffic_realtime) \ .trigger(processingTime1 minute) \ .start()withWatermark设 2 分钟是容忍乱序的边界交通数据从设备上报到进 Kafka 一般延迟在秒级2 分钟足够覆盖大部分乱序。outputMode(update)只输出变化的窗口比complete省资源。trigger设 1 分钟意味着微批间隔延迟和吞吐的折中点。checkpoint 目录必须放在可靠存储上否则重启后状态丢失会重复计算。这里有个容易忽略的点流处理的聚合结果和离线结果对不上是正常的因为窗口边界和乱序处理策略不同。我的习惯是在报表层标注数据来源实时值用于监控告警离线值用于正式统计两者不混用。另外流式写入 parquet 会产生大量小文件生产环境一般换成 Delta 或 Hudi课程设计阶段用 parquet 加定时合并也能接受。调优上流处理对 executor 数量不敏感但对内存敏感因为状态要常驻。spark.sql.streaming.stateStore.providerClass默认的 HDFSBackedStateStore 在状态大时性能一般数据量上来后可以换 RocksDB。这些参数不用一开始就调先让链路跑通压测时再逐个改。最后说个我自己的习惯每次改完聚合逻辑先拿一天的历史数据用批处理跑一遍确认指标口径对了再切到流。批流用同一套 SQL能省掉大量对账时间。这套系统值不值得做取决于你手上有没有持续产生的交通数据——有数据Spark 的分层和聚合能力能让你从「导出一张 Excel」变成「随时查任意路段任意时段」没数据先把公开的卡口数据集跑通链路再考虑接真实源。希望帮到你。本文还有配套的精品资源点击获取