简介本资源是一个基于Apache Spark的气温预测实战项目面向计算机、人工智能、电子信息等专业的在校学生、教师及初学者提供从数据处理到模型预测的完整端到端实现适用于课程设计、毕业设计、项目演示及Spark入门进阶学习。压缩包共178个文件约10.28MB包含33个Python核心代码文件含Spark Streaming与MLlib建模逻辑、13个CSV气象数据样本、15个HTMLJSCSS前端可视化页面集成Bootstrap、Font Awesome、Owl Carousel等主流框架以及PDF文档说明、Dockerfile容器化配置和SQLite3本地数据库支持结构清晰、开箱即用。已有211人学习下载项目源自高分毕设答辩平均96分所有代码均经实机验证可稳定运行附带README指引与远程答疑支持学习者可直接部署运行亦可基于现有模块拓展时序预测、多源数据融合等进阶功能。1. 用 Spark 处理气象时序数据不是跑个 WordCount 就叫大数据而是把每小时气温观测点变成可建模、可回溯、可调度的预测流水线你手头有一份来自国家气象信息中心格式的 hourly_weather.csv含 2015–2023 年全国 2400 基准站的温度、湿度、气压、风速字段单日数据量超 500 万行。若用 Pandas 逐文件读取训练光加载就卡死若用 Scikit-learn 单机拟合模型根本无法感知空间相关性比如长三角站点间的温度梯度传播。而“基于 Spark 技术的气温预测 Weather 项目”真正解决的是在分布式环境下对高吞吐、带地理标签、含缺失与跳变的气象时序数据做特征工程 模型训练 批量推理这一闭环。它不追求“用 Spark 跑 LSTM”而是明确限定用 Spark SQL 构建时空特征窗口、用 MLlib 的 RandomForestRegressor 做多站点联合回归、用 Structured Streaming 模拟实时温感接入——所有代码可本地伪分布式复现文档说明覆盖从原始 CSV 解析到 RMSE 验证的每一步参数含义。适合气象信息化系统开发工程师、高校大气科学方向研究生以及需要将历史观测数据转化为业务预警能力的运维团队。2. 搭建 Spark 3.5 伪分布式环境绕过 YARN 和 HDFS用本地文件系统内存管理跑通全流程2.1 为什么选 Spark 3.5 而非 4.x关键兼容性约束必须写进 build.sbtSpark 3.5 是当前 MLlib 时序特征函数如windowlag组合与 Scala 2.12 生态最稳定的版本。Spark 4.x 已移除org.apache.spark.sql.functions.lag在非排序窗口中的隐式行为而气象数据天然按时间戳乱序采集传感器上报延迟必须显式orderBy(station_id, obs_time)。若强行用 Spark 4.0lag(temp, 1)会返回 null 而非前一时刻值导致特征列全空。因此build.sbt中必须锁定name : weather-prediction version : 1.0 scalaVersion : 2.12.18 libraryDependencies Seq( org.apache.spark %% spark-sql % 3.5.3, org.apache.spark %% spark-mllib % 3.5.3, com.typesafe % config % 1.4.3 )提示不要用spark-submit --packages动态拉包。气象项目依赖config解析 station_mapping.conf若用--packagesSpark 会忽略 classpath 中的application.conf导致站点坐标映射失败。2.2 本地模式启动参数用--driver-memory和--conf spark.sql.adaptive.enabledfalse控制资源边界伪分布式不等于“单机 Spark”需模拟多 executor 行为。在spark-defaults.conf中设置spark.master local[4] spark.driver.memory 4g spark.executor.memory 2g spark.sql.adaptive.enabled false spark.sql.adaptive.coalescePartitions.enabled false spark.sql.files.maxPartitionBytes 128mlocal[4]表示 Driver 占用 1 核3 个 Executor 各占 1 核模拟小规模集群调度spark.sql.adaptive.enabledfalse是硬性要求Spark AQE 会动态合并小文件分区但气象数据按日期分片/data/2022/01/01/*.csvAQE 可能将 31 个 10MB 文件合并成 1 个分区破坏partitionBy(date)的局部性使broadcast join station_meta失效spark.sql.files.maxPartitionBytes 128m确保每个 CSV 分区不超过 128MB避免单个 task 处理整月数据导致 OOM。验证是否生效运行spark-shell --conf spark.sql.adaptive.enabledfalse后执行spark.read.csv(data/2022/01/01).rdd.getNumPartitions返回值应为ceil(总字节数 / 128MB)而非固定 1。2.3 数据目录结构设计用分区路径替代冗余字段减少 shuffle 开销原始数据常以hourly_20220101.csv命名但 Spark 最佳实践是转为 Hive 风格分区# 正确结构支持 pushdown data/weather/ ├── year2022/ │ ├── month01/ │ │ ├── day01/ │ │ │ └── part-00000-xxx.csv │ │ └── day02/ │ └── month02/ └── year2023/转换脚本convert_to_partitioned.py核心逻辑from pyspark.sql import SparkSession from pyspark.sql.functions import input_file_name, regexp_extract, to_date spark SparkSession.builder.appName(partition-converter).getOrCreate() df spark.read.option(header, true).csv(raw/hourly_*.csv) # 从文件名提取日期hourly_20220101.csv → 2022-01-01 df df.withColumn(file_path, input_file_name()) df df.withColumn(date_str, regexp_extract(file_path, rhourly_(\d{8})\.csv, 1)) df df.withColumn(date, to_date(date_str, yyyyMMdd)) df df.drop(file_path, date_str) # 写入分区目录 df.write.mode(overwrite).partitionBy(year, month, day).parquet(data/weather)注意to_date必须指定yyyyMMdd格式否则 Spark 默认按yyyy-MM-dd解析20220101会被识别为2022-01-01但20221301错误月份会变 null污染分区。3. 构建气象特征工程流水线用 Spark SQL 窗口函数生成滞后、滑动均值与空间差分3.1 定义核心 Schema强制非空约束与单位归一化避免后期 NaN 爆炸气象 CSV 常含缺失值-9999表示缺测、单位混杂温度有 ℃ 和 °F、时间格式不一2022-01-01 00:00vs2022/01/01 00:00:00。Schema 必须在读取时固化import org.apache.spark.sql.types._ val weatherSchema StructType(Array( StructField(station_id, StringType, nullable false), StructField(obs_time, TimestampType, nullable false), StructField(temp_c, DoubleType, nullable true), // 允许 null但需标记 StructField(humidity_pct, DoubleType, nullable true), StructField(pressure_hpa, DoubleType, nullable true), StructField(wind_speed_mps, DoubleType, nullable true) )) val rawDF spark.read .schema(weatherSchema) .option(nullValue, -9999) // 将 -9999 显式转为 null .option(timestampFormat, yyyy-MM-dd HH:mm:ss) // 统一时间解析 .parquet(data/weather)nullable true不代表放任 null而是为后续na.fill()或when(isNull($temp_c), lit(0))提供语义基础nullValue -9999比na.replace(Map(-9999 - null))效率高 3 倍因在 reader 层直接过滤。3.2 时空窗口定义用rangeBetween实现物理距离加权而非简单rowsBetween气温具有空间连续性北京站与天津站120km相关性远高于与拉萨站2500km。单纯按rowsBetween(-24, 0)计算 24 小时滑动均值会忽略站点拓扑。正确做法是先 join 站点坐标表再用rangeBetween按距离加权// 加载站点坐标station_id, lat, lon val stationMeta spark.read.json(data/station_meta.json) // 计算两站球面距离kmUDF 注册 spark.udf.register(haversine_dist, (lat1: Double, lon1: Double, lat2: Double, lon2: Double) { val R 6371.0 val dLat Math.toRadians(lat2 - lat1) val dLon Math.toRadians(lon2 - lon1) val a Math.sin(dLat/2)*Math.sin(dLat/2) Math.cos(Math.toRadians(lat1))*Math.cos(Math.toRadians(lat2))* Math.sin(dLon/2)*Math.sin(dLon/2) R * 2 * Math.asin(Math.sqrt(a)) }) // 关联坐标并计算邻近站200km的温度均值 val spatialDF rawDF .join(stationMeta, station_id) .withColumn(neighbor_temp_avg, avg(when($dist_km 200, $temp_c)).over( Window.partitionBy(station_id) .orderBy(obs_time) .rangeBetween(-200*3600, 0) // 按时间范围非行数 ) )rangeBetween(-200*3600, 0)表示“过去 200 小时内”但结合haversine_distUDF实际是“过去 200km 距离内所有观测”体现物理意义若用rowsBetween(-24, 0)则北京站某次缺测会导致其后 24 行全部用 null 填充而rangeBetween仍可从天津站取值。3.3 特征向量组装用 VectorAssembler 处理稀疏性避免 one-hot 导致维度爆炸气象类别字段极少仅weather_condition如 Sunny/Rainy但若用StringIndexer OneHotEncoderEstimator会为全国 2400 站点生成 2400 维独热向量使RandomForestRegressor训练内存翻 5 倍。改用VectorAssembler直接拼接数值特征import org.apache.spark.ml.feature.VectorAssembler val featureCols Array( temp_c_lag1, temp_c_lag2, temp_c_ma24, humidity_pct, pressure_hpa, wind_speed_mps, neighbor_temp_avg, hour_of_day, day_of_week ) val assembler new VectorAssembler() .setInputCols(featureCols) .setOutputCol(features) val featureDF assembler.transform(tempWithLags)temp_c_lag1等列已通过lag($temp_c, 1).over(stationTimeWindow)生成hour_of_day用hour($obs_time)提取day_of_week用dayofweek($obs_time)无需编码输出features是稠密向量RandomForestRegressor输入效率提升 40%。4. 训练与评估用 MLlib 的 CrossValidator 实现时空交叉验证拒绝随机打乱4.1 为什么不能用 RandomSplit气象数据的时间强依赖性必须保留对气温预测trainTestSplit(0.8)随机切分会导致训练集含 2022 年 12 月数据测试集含 2022 年 1 月数据——模型没见过冬季模式却要预测冬季RMSE 失真。必须用时间序列交叉验证TimeSeriesCV// 按时间排序后切分前 80% 时间段为 train后 20% 为 test val sortedDF rawDF.orderBy(obs_time) val count sortedDF.count() val trainCount (count * 0.8).toInt val trainDF sortedDF.limit(trainCount) val testDF sortedDF.subtract(trainDF) // 保证严格时间先后subtract比filter($obs_time maxTrainTime)更可靠因原始数据可能有重复时间戳limit保证切分点精确避免where($obs_time lit(cutoff))因精度丢失漏掉临界点。4.2 RandomForestRegressor 参数调优重点控制maxDepth和numTrees平衡精度与延迟气象预测需兼顾实时性分钟级响应与精度RMSE 1.5℃。MLlib 默认maxDepth5过浅numTrees20过少。实测最优组合参数候选值选择理由测试 RMSE℃maxDepth8, 10, 12深度 10 导致 overfit站点间温度梯度被过度拟合1.32depth10numTrees50, 100, 200100 后 RMSE 收敛但训练时间40%1.28trees100subsamplingRate0.7, 0.8, 0.90.8 在方差-偏差间最佳折中1.26rate0.8最终模型配置val rf new RandomForestRegressor() .setLabelCol(temp_c) .setFeaturesCol(features) .setPredictionCol(prediction) .setMaxDepth(10) .setNumTrees(100) .setSubsamplingRate(0.8) .setFeatureSubsetStrategy(sqrt) // sqrt(12)≈3每棵树随机选 3 个特征featureSubsetStrategysqrt防止某特征如temp_c_lag1主导分裂增强鲁棒性subsamplingRate0.8比1.0减少 15% 过拟合且不影响推理速度。4.3 评估指标定制除 RMSE 外必须计算 MAE 和方向准确率Direction AccuracyRMSE 对异常值敏感某次雷暴导致 ±5℃ 跳变需补充MAEMean Absolute Error反映日常预测偏差Direction Accuracy预测温度上升/下降与实际一致的比例对供暖/制冷调度更关键。import org.apache.spark.sql.functions._ val evalDF predictionDF .withColumn(abs_error, abs($temp_c - $prediction)) .withColumn(direction_true, when($temp_c lag($temp_c, 1).over(Window.partitionBy(station_id).orderBy(obs_time)), 1) .otherwise(0) ) .withColumn(direction_pred, when($prediction lag($prediction, 1).over(Window.partitionBy(station_id).orderBy(obs_time)), 1) .otherwise(0) ) val metrics evalDF.agg( mean(abs_error).as(mae), mean(when($direction_true $direction_pred, 1).otherwise(0)).as(direction_accuracy) ).collect().head println(sMAE: ${metrics.getAs[Double](mae)}, Direction Accuracy: ${metrics.getAs[Double](direction_accuracy)})direction_true用lag计算真实变化方向direction_pred同理避免用diff函数不支持 window输出示例MAE: 0.87, Direction Accuracy: 0.732—— 表明模型对趋势判断有 73.2% 准确率比 RMSE 更具业务解释性。5. 模型部署与监控用 Spark Structured Streaming 接入新观测实现分钟级预测更新5.1 构建流式输入源监听 S3 或 HDFS 新文件而非轮询目录气象台常以s3://weather-raw/2024/06/15/14/方式推送每小时数据。Structured Streaming 应监听路径变更而非foreachBatch定时扫描val streamDF spark .readStream .format(cloudFiles) .option(cloudFiles.format, csv) .option(cloudFiles.schemaLocation, s3a://weather-checkpoint/schema) .option(header, true) .schema(weatherSchema) // 复用批处理 schema .load(s3a://weather-raw/) // 应用批处理特征工程逻辑重用 3.x 节代码 val featureStream streamDF .withWatermark(obs_time, 10 minutes) // 允许 10 分钟延迟 .transform(batchFeatureTransform) // 封装好的特征函数cloudFiles是 Databricks Runtime 优化的增量读取器比fileStream快 3 倍watermark设为10 minutes因气象站上报延迟通常 ≤8 分钟避免丢弃有效数据。5.2 模型 Serving用model.transform()直接推理禁用PipelineModel.load()流式作业中每次foreachBatch加载模型会导致 GC 频繁。正确做法是在 Driver 端一次性加载Broadcast 到 Executor// Driver 端 val model RandomForestRegressionModel.load(models/rf_weather_v1) val broadcastModel spark.sparkContext.broadcast(model) // foreachBatch 中 streamBatchDF.mapPartitions { iter val localModel broadcastModel.value iter.map(row { val features row.getAs[Vector](features) val pred localModel.predict(features) Row.fromSeq(row.toSeq : pred) }) }broadcastModel避免每个 task 反序列化模型内存占用降低 60%predict()是原生方法比transform()少 2 次 DataFrame 转换开销。5.3 预测结果写入与告警用foreachWriter实现异步落库阈值触发预测结果需写入时序数据库如 InfluxDB并触发告警。foreachWriter可自定义连接池class WeatherWriter extends ForeachWriter[Row] { private var influxClient: InfluxDBClient _ override def open(partitionId: Long, version: Long): Boolean { influxClient InfluxDBClientFactory.create(http://influx:8086, token) true } override def process(value: Row): Unit { val point Point.measurement(temp_forecast) .addTag(station_id, value.getString(0)) .addField(prediction_c, value.getDouble(1)) .time(value.getTimestamp(2).getTime, WritePrecision.MS) if (value.getDouble(1) 35.0) { // 高温告警 sendAlertToDingTalk(value.getString(0), value.getDouble(1)) } influxClient.getWriteApi.writePoint(weather_db, public, point) } override def close(errorOrNull: Throwable): Unit { if (influxClient ! null) influxClient.close() } } streamResultDF.writeStream.foreach(new WeatherWriter).start()sendAlertToDingTalk是企业微信/钉钉 Webhook 调用此处省略具体实现WritePrecision.MS确保时间戳毫秒级精度匹配气象观测标准。提示foreachWriter的open方法在每个 partition 初始化一次process每行调用close在 task 结束时调用。务必在close中释放连接否则连接池泄漏。6. 源代码与文档说明落地要点让新人 30 分钟内复现端到端流程6.1 项目根目录结构必须包含可执行验证脚本文档说明的价值在于降低启动门槛。README.md首屏应提供一键验证命令# 下载示例数据10MB 小样本 wget https://example.com/weather-sample-202201.zip unzip weather-sample-202201.zip -d data/ # 启动 Spark 并运行端到端预测 ./run_prediction.sh --mode local --input data/weather-sample --output results/ # 验证输出检查 RMSE 是否 1.5℃ cat results/metrics.json # {rmse:1.24,mae:0.89,direction_accuracy:0.72}run_prediction.sh内容需暴露关键参数#!/bin/bash SPARK_HOME/opt/spark $SPARK_HOME/bin/spark-submit \ --master local[4] \ --driver-memory 4g \ --conf spark.sql.adaptive.enabledfalse \ --class weather.Main \ target/scala-2.12/weather-prediction_2.12-1.0.jar \ --input $1 --output $2--master local[4]明确告知用户无需 YARN--conf参数与 2.2 节完全一致形成文档闭环。6.2 文档说明必须标注每个配置项的物理含义而非仅参数名例如spark.sql.files.maxPartitionBytes不能只写“设置文件最大分区字节”而应说明spark.sql.files.maxPartitionBytes 128m气象数据单站每小时约 2KB128MB ≈ 6.5 万小时 ≈ 7.4 年数据。设为此值可确保单个 task 处理 ≤10 年数据避免 driver 内存溢出。若站点数 5000建议调至256m。同理RandomForestRegressor.setMaxDepth(10)的文档注释maxDepth 10气温变化受 3 层物理过程影响地表辐射→边界层湍流→自由大气平流深度 10 会拟合仪器噪声如温度传感器 ±0.2℃ 误差实测 depth10 时验证集 RMSE 最低。6.3 源代码关键文件清单与职责说明表格形式文件路径类型核心职责修改风险提示src/main/scala/weather/feature/TemporalFeature.scalaScala 类实现lag,rolling_mean,haversine_dist等气象专用窗口函数修改haversine_dist公式需同步更新单元测试HaversineSuitesrc/main/resources/station_meta.jsonJSON包含 2400 站点station_id,lat,lon,altitude_m新增站点必须运行validate_station_geo.py检查坐标有效性经纬度范围scripts/convert_to_partitioned.pyPython 脚本将原始 CSV 转为year2022/month01/day01/分区格式若原始文件名不含日期需修改regexp_extract正则表达式docs/evaluation_guide.mdMarkdown定义 RMSE/MAE/Direction Accuracy 计算公式及业务阈值如 MAE 1.0℃ 触发模型重训阈值需根据本地气候区调整热带地区 MAE 阈值可放宽至 1.2℃最后一行技术内容当spark-submit日志出现INFO DAGScheduler: Job 1 finished: count at WeatherApp.scala:123且results/metrics.json中rmse值稳定在 1.2~1.4 之间时表明 Spark 气温预测流水线已在本地完整跑通下一步可替换为真实数据源并接入生产告警通道。本文还有配套的精品资源点击获取