★ FEATURED ARTICLE 先说结论这个项目真正的难点不在“预测”而在把 Hadoop 和 PySpark 这套分布式处理逻辑跟 Prophet 这种以 pandas 数据框为核心的单机模型舒服地拼在一起。我前后折腾了大概两周踩的坑比写的代码还多今天把完整的思路、代码片段和排错过程都整理出来希望能帮后来的人少走点弯路。这个系统适合谁如果你是第一次接触“大数据 时间序列预测”这种组合或者已经单机跑通过 Prophet 但不知道怎么往 Spark 上迁这篇内容可以给你一条完整的参考路径。文章里涉及的操作我都以模拟项目X一套城市空气质量历史数据为例所有字段和参数都做了脱敏处理但逻辑完全可复现。1. 为什么非要把 Hadoop、PySpark 和 Prophet 攒在一起1.1 项目背景单机脚本撑不住之后最初这个需求特别朴素把某城市过去五年的空气质量监测数据拿出来预测未来七天的 PM2.5。数据量听起来也不夸张一年大概几十万条记录五年也就两三百万行。用 pandas 读 CSV 再喂给 Prophet理论上完全扛得住。但实际一跑就发现问题了——原始数据散落在十几个不同来源里有的是按小时上报的站点传感器读数有的是天气站下载的气象表还有一部分是第三方接口导出的日志文件格式、字段名、时间粒度全都不一样。我第一版写了个纯 pandas 脚本处理流程就是“读一个文件 - 清洗一个文件 - 合并到一个 DataFrame”跑一次全量清洗大概要四十多分钟。而且每次新增数据源或者调整过滤逻辑就得从头再来一遍。最难受的是内存某几次加载全量历史数据时直接把开发机的 16G 内存吃满连 IDE 都开始卡。到这一步我意识到问题不是 Prophet 太慢而是“数据准备阶段”缺少一个能横向扩展的清洗层。这时候引入 Hadoop 和 PySpark 的目的就清晰了HDFS 负责把分散的文件统一收拢存储PySpark 负责把清洗和特征工程变成分布式任务。原始文件丢到 HDFS 上spark 作业并行读、并行洗、并行算跑一次全量清洗从四十多分钟降到五六分钟而且以后加节点就能继续扩。1.2 技术选型的真实理由哪些环节真的需要 Spark很多人一听“Hadoop Spark 预测”下意识觉得这是杀鸡用牛刀。我实际用下来整套栈里的分工是这样的HDFS解决“文件散落 历史数据归档”问题。所有原始 CSV 统一放到 HDFS 指定目录后续清洗任务只要能访问 HDFS 就能处理不依赖某台机器上的本地文件。PySpark清洗阶段解决“多数据源 join 和重复清洗”问题。不同站点、不同时间粒度的数据用 DataFrame 的groupBy pivot join在分布式环境里跑比 pandas 的merge要稳当得多尤其是内存不再成为瓶颈。Prophet预测阶段真正跑模型的部分依然是单机。Prophet 本身没有分布式实现它需要的输入是一张规整好的、包含ds和y两列的 DataFrame这个数据量在模型训练时可大可小但常规城市级数据几十万行规模单机完全跑得动。所以准确的讲这是一个“分布式数据处理 单机模型训练”的混合架构。Spark 不负责预测只负责把数据喂到 Prophet 嘴边的过程。明确这一点后面所有设计都会顺很多。1.3 总体数据流向我最后搭出来的数据流向是这样的原始文件多源 CSV/JSON - HDFS 落盘 - PySpark 清洗作业 - 特征宽表按小时粒度 - 导出成建模所需 CSV - Prophet 训练与预测 - 预测结果写入结果表 - 可视化服务读取并展示这个链条里只有最后一个“可视化服务读取”用的是传统关系库或 ES中间清洗结果我直接用了 Parquet 格式存 HDFS建模前再导出成临时 CSV。这样做的好处是中间产物可回溯、可重跑不会因为某一步参数调错就把源头数据污染了。2. 环境准备版本匹配才是第一个隐形杀手2.1 版本组合与 Java/Spark 兼容这个项目最大的坑在环境依赖。我一开始随手装了最新版 Spark结果跟 Hadoop 的 RPC 协议对不上作业提交后一直报Connection refused。后来梳理了一遍兼容矩阵确定了一套稳定组合组件版本说明JavaOpenJDK 8Spark 3.x 对 Java 8 支持最好Java 11 也能跑但个别扩展包有兼容问题Hadoop3.3.4只用到 HDFS 和 YARN 基础能力Spark3.2.1编译时对应 Hadoop 3.2 以上即可实际测试 3.3.4 无问题Python3.9太高的版本比如 3.11在某些 Prophet 依赖上编译会有麻烦Prophet1.1.4这个版本对 pandas 1.5.x 兼容性最好这里单独说一下 Python 版本。Prophet 底层依赖 cmdstanpy新版 Stan 对编译器要求高如果你用的是 Python 3.11 或 3.12在部分 Linux 发行版上安装 prophet 时会卡在编译 C 扩展那一步。我为了省事直接把环境固定在 Python 3.9 Prophet 1.1.4整个安装过程一条pip install prophet就能过不折腾。2.2 PySpark 内存与 YARN 配置PySpark 作业跑在 YARN 上时最常遇到的问题是Container killed on request或GC overhead limit exceeded。这个不一定是代码写错了更多是资源参数没配好。我的开发机是 8 核 16G配置如下# spark-defaults.conf 关键参数 spark.masteryarn spark.driver.memory2g spark.executor.instances3 spark.executor.memory3g spark.executor.cores2 spark.yarn.am.memory1g spark.sql.shuffle.partitions50 spark.default.parallelism50spark.sql.shuffle.partitions这个参数很容易被忽略。默认值是 200也就是说执行 join 或 groupBy 的时候会默认生成 200 个分区。对这份数据量来说200 个分区只会徒增调度开销每个分区里又没多少数据反而拖慢速度。调成 50 之后整个清洗作业快了将近一倍。如果你是在本地单机跑 spark-submit不想起 YARN 集群可以直接用spark.masterlocal[4]但要注意local[*]的星号有时会拿到所有核跟其他任务抢资源。2.3 开发模式选择本地调试加集群提交我的建议是分成两个模式。日常写 PySpark 清洗逻辑时直接在 IDE 里用local[2]模式跑配上少量采样数据做验证这样报错时堆栈信息是全的非常方便确认逻辑没问题了再把作业提交到 YARN 上跑全量数据。通过在 spark session 里加一个全局开关来控制运行模式即可import os RUN_MODE os.getenv(RUN_MODE, local) # local / cluster if RUN_MODE local: spark SparkSession.builder \ .appName(AirQualityClean_Local) \ .master(local[2]) \ .config(spark.sql.shuffle.partitions, 20) \ .getOrCreate() else: spark SparkSession.builder \ .appName(AirQualityClean_Cluster) \ .master(yarn) \ .config(spark.executor.memory, 3g) \ .config(spark.sql.shuffle.partitions, 50) \ .getOrCreate()这样最舒服的一点是写逻辑和写配置分离。我在本地改了清洗逻辑直接跑小数据集验证确认无误后把同一份代码用--master yarn方式提交到集群源代码完全不用改。3. 数据接入与清洗从 HDFS 原始文件到可建模宽表3.1 数据字段与口径统一空气质量监测的原始数据最常见的字段大致有这些站点编号、监测时间、PM2.5 浓度、PM10 浓度、SO2 浓度、NO2 浓度、O3 浓度、CO 浓度外加气象侧的温度、湿度、气压、风速、降水等。不同来源的文件里时间格式有的叫time有的叫record_time有的是字符串2023-01-01 08:00:00有的是 Unix 时间戳。第一步必须统一列名和时间格式。我统一的口径是时间列统一命名为time格式为yyyy-MM-dd HH:00:00全部按整点对齐。浓度单位统一为 μg/m³CO 为 mg/m³。30 分钟内重复的记录保留后一条。空气质检数据若某小时缺失用前后两小时均值插值前提是连续缺失不超过 3 小时超过 3 小时则标记为null在建模阶段由 Prophet 自动处理。统一时间格式这一步是后面所有 join 操作的基础。不同数据源的时间精度如果不一致比如一个精确到分一个精确到秒直接 join 会产生大量无法匹配的行。我建议哪怕原始文件里是精确到秒的时间也先截断成整点再合并。3.2 读取 HDFS 原始文件的 PySpark 写法把文件考到 HDFS 之后读取逻辑非常简单。用 Spark 原生 CSV 读取器加上 Schema 定义可以避免自动推断类型带来的性能损耗from pyspark.sql.types import StructType, StructField from pyspark.sql.types import StringType, DoubleType, TimestampType schema StructType([ StructField(station_id, StringType(), True), StructField(record_time, StringType(), True), StructField(pm25, DoubleType(), True), StructField(pm10, DoubleType(), True), StructField(so2, DoubleType(), True), StructField(no2, DoubleType(), True), StructField(o3, DoubleType(), True), StructField(co, DoubleType(), True), ]) df_raw spark.read.csv( hdfs:///user/airquality/raw/station_*.csv, headerTrue, schemaschema, sep, )这里有个小细节文件路径用通配符station_*.csvSpark 会自动把匹配到的多个文件当作同一张表来读。如果你某个数据源的文件特别多比如按天分文件存储这种写法比逐个文件列出路径高效太多。3.3 清洗逻辑数值过滤比想象中复杂原始数据里最坑的是异常值。空气监测仪在故障或校准时会产生一些极端值比如 PM2.5 瞬间变成 999、-1、或者 0。直接删掉是不合适的因为有部分 0 是真实值雨后极洁净天气但 999 这种一般是传感器故障。我的处理策略分两层硬过滤浓度值小于 0 的删除PM2.5、PM10 大于 500 的删除极端沙尘暴除外但本项目范围内先按此逻辑。区间校验用 3σ 原则识别离群点但只标记不删除通过把离群值按前后 6 小时均值平滑处理。这一层在 PySpark 里可以用自定义 UDF 或者窗口函数实现。推荐用窗口函数性能比 UDF 好得多from pyspark.sql.window import Window from pyspark.sql.functions import avg, col, when w Window.partitionBy(station_id).orderBy(time).rowsBetween(-3, 3) df_clean df_raw \ .withColumn(pm25_smooth_avg, avg(pm25).over(w)) \ .withColumn(pm25, when(abs(col(pm25) - col(pm25_smooth_avg)) 150, col(pm25_smooth_avg)) .otherwise(col(pm25)))简单说就是把每个站点前后 3 小时共 7 个点的平均值算出来如果某个时刻的原始值跟这个平均值的差超过 150就把这个点替换成平均值。这种方法对单点突变非常有效而且不会把真实的高浓度污染过程抹掉——因为真实污染过程是持续数小时的不会只突变一个点。3.4 多站点汇总与气象表合并城市空气质量预测通常要的是“全市平均”或“重点站点平均”而不是把每个站点单独训练一个模型。我这里用的是简单平均法所有站点同一小时的数据求均值得到逐小时全市均值。同时把气象数据表 join 过来。df_station_avg df_clean \ .groupBy(time) \ .agg( avg(pm25).alias(pm25), avg(pm10).alias(pm10), avg(so2).alias(so2), avg(no2).alias(no2), avg(o3).alias(o3), avg(co).alias(co) ) df_weather_avg df_weather \ .groupBy(time) \ .agg( avg(temp).alias(temp), avg(humidity).alias(humidity), avg(pressure).alias(pressure), avg(wind_speed).alias(wind_speed) ) df_feature df_station_avg.join(df_weather_avg, ontime, howleft)这里 join 用left而不是inner是因为气象数据偶发缺失缺失时段后续由 Prophet 内部处理或简单前向填充直接影响不大。3.5 导出建模数据清洗完之后我通常会把结果存成两份一份是 Parquet 格式放到 HDFS 上作为中间层可回溯一份是导出成 CSV 放到本地供 Prophet 读取。导出用 PySpark 的repartition(1)确保只生成一个文件避免后续读取时多文件拼接df_feature.repartition(1) \ .write.mode(overwrite) \ .csv(/user/airquality/clean/air_feature_hourly.csv, headerTrue)然后用hadoop fs -get拉到本地即可。这一步有个经验repartition(1)在数据量大时会有一定性能损耗但从建模角度讲只有一个文件会让 Prophet 侧读取简单很多两三百 MB 的数据量完全能接受。4. Prophet 模型构建为什么默认参数在空气质量数据上会翻车4.1 Prophet 适用于空气质量的原理Prophet 的本质是“可分解加性模型”把时间序列分解成趋势项、季节项、节假日效应项和误差项。对空气质量数据来说这个分解天然合理PM2.5 浓度有明显的年季周期冬季高、夏季低、有日内周期早晚高峰高、午后低还有由天气突变带来的非周期波动冷空气过境后的扩散条件变化。这些正好是 Prophet 的舒适区。输入 Prophet 的数据只需要两列ds时间戳必须是 datetime 类型。y目标值数值类型。其他所有特征——比如温度、湿度、风速——如果想纳入模型需要用add_regressor加为额外回归项。4.2 调参关键季节模式和变点尺度我第一版直接用默认参数跑结果极其拉胯预测曲线几乎是一条纯趋势线完全没抓住周期波动MAPE 高达 40% 以上。问题出在两个默认参数上seasonality_modeadditive空气质量数据里冬季 PM2.5 可能到 150夏季可能只有 30这时候季节效应大小跟数值本身的量级是成比例的需要用乘法模式。changepoint_prior_scale0.05默认值偏保守导致模型不敢在数据里识别突变点。空气质量受政策管控、产业调整、季节转换影响平台期和后期的结构突变不少需要适当放开。我最后调出来的核心参数组合from prophet import Prophet model Prophet( seasonality_modemultiplicative, changepoint_prior_scale0.1, seasonality_prior_scale10.0, yearly_seasonalityTrue, weekly_seasonalityTrue, daily_seasonalityTrue, holidays_prior_scale15.0 ) # 添加气象回归变量 model.add_regressor(temp) model.add_regressor(humidity) model.add_regressor(pressure) model.add_regressor(wind_speed) model.fit(df_train)补充说明一下seasonality_prior_scale的作用它控制季节效应的灵活度。默认值是 10但在我这份数据上默认值表现出的季节分量偏“平”提高一点能让模型更好地拟合夏季低谷和冬季峰值。这不是说越大越好太大容易过拟合需要配合交叉验证去选。4.3 训练集与评估指标选用近三年的数据做训练近 30 天做验证预测未来则是 7 天。评估指标用三个MAE平均绝对误差反映整体偏差。RMSE均方根误差对大幅偏差更敏感。MAPE平均绝对百分比误差这块数据的业务上比较看重相对误差。用 Prophet 自带的cross_validation做时间序列交叉验证from prophet.diagnostics import cross_validation, performance_metrics import pandas as pd df_cv cross_validation( model, initial730 days, period30 days, horizon7 days ) df_metrics performance_metrics(df_cv) print(df_metrics[[mae, rmse, mape]].head())在我这份模拟数据上的结果大致是MAE 在 12~15 μg/m³ 区间MAPE 控制在 20% 上下。作为对比纯线性回归做同样的预测 MAPE 要到 40%随机森林效果介于两者之间但周期外推能力差不少。Prophet 的强项就在于“没有未来气象预报数据时单靠历史周期也能给出不差的基准线”。4.4 预测未来 7 天并输出结果表预测未来 7 天时需要构造一个future数据框同时给未来 7 天的气象回归变量填入历史同期均值或简单预测值future model.make_future_dataframe(periods7, freqH) # 未来 7 天温度用近 5 年同日同时刻平均值填充 future future.merge(future_weather, howleft, onds) forecast model.predict(future) forecast[[ds, yhat, yhat_lower, yhat_upper]].tail(24 * 7)yhat_lower和yhat_upper是置信区间可视化时非常有用能清楚地展示“预测值的可信范围”。如果区间宽度过大比如超过 50 μg/m³说明模型对该时段信心不足业务上就要谨慎解读。5. 预测结果可视化不只是画线那么简单5.1 静态图先看趋势是否合理预测结果出来后第一件事是画一张静态图确认模型行为合理。Prophet 自带的 plot 方法是最快的检查手段fig1 model.plot(forecast) fig2 model.plot_components(forecast)plot_components会画出趋势项、周季节项、日季节项三个子图。这一步非常值得花时间看它能帮你确认模型是不是学到了合理规律日均季节项应该有早晚两个峰周季节项如果是“周末低、工作日高”就很合理车辆排放减少年季节项应该是冬高夏低。我就在这张图上发现过一个问题日季节项在凌晨 4 点出现了一个莫名其妙的凸起查了半天发现是某几个站点夜间的仪器零点校准导致数据系统性偏高这属于数据问题而非模型问题。如果没有组件分解图这种异常很难快速定位。5.2 交互式可视化pyecharts 是最好用的选择静态图自用可以但如果要给非技术背景的人看或者部署成内部页面推荐用 pyecharts 做交互式折线图。它渲染出的 HTML 页面无需后端服务直接打开就能用也方便嵌入内部系统。from pyecharts.charts import Line from pyecharts.options import TooltipOpts, TitleOpts line Line(init_opts{width: 1200px, height: 500px}) line.add_xaxis(actual[ds].dt.strftime(%m-%d %H:00).tolist()) line.add_yaxis( 实际值, actual[pm25].tolist(), is_smoothTrue, linestyle_opts{width: 2}, itemstyle_opts{color: #333} ) line.add_yaxis( 预测值, forecast[yhat].tolist(), is_smoothTrue, linestyle_opts{width: 2, type: dashed}, itemstyle_opts{color: #d14a22} ) line.add_yaxis( 置信上界, forecast[yhat_upper].tolist(), linestyle_opts{type: dotted} ) line.add_yaxis( 置信下界, forecast[yhat_lower].tolist(), linestyle_opts{type: dotted} ) line.set_global_opts( title_optsTitleOpts(title未来7天 PM2.5 浓度预测), tooltip_optsTooltipOpts(triggeraxis) ) line.render(air_quality_forecast_7d.html)这里我的经验是把“实际值”“预测值”“置信上界”“置信下界”四条线放在同一张图里。实际值展示的是最近 48 小时的真实数据预测值展示未来 7 天的预测这样看图的人能很清楚看到模型从哪个位置开始切换成预测模式也容易判断预测合理性。5.3 大屏展示盯住“拐点”而不是“数值”如果要做成值班大屏我不建议把大量图表堆在一起。真正对决策有用的信息就三块未来 7 天预测曲线重点看趋势拐点是持续升高还是持续下降。未来 24 小时的逐小时精度预测表重点看红线阈值时刻。模型置信区间范围判断预测是否稳定。大屏的配色也别花哨用深色底、亮色线信号要突出“拐点”——比如未来 48 小时出现浓度持续上升主题色从绿色跳成橙色。这是做可视化时特别容易忽略的线图展示的是整体走势而业务方每天盯的是“哪天会转差”不看绝对数值。6. 实际运行中的坑与注意事项6.1 PySpark 清洗与 Prophet 建模之间的数据对齐问题这是我这次项目里最折磨人的一个问题。PySpark 清洗完导出的 CSV时间列默认会带有00:00或08:00这种时区偏移后缀。Prophet 在读取这类字符串时有时间能自动解析有时间会报ValueError: Unable to parse。解决办法是在 PySpark 导出前统一转成无时区的字符串格式from pyspark.sql.functions import date_format df_feature df_feature.withColumn( ds, date_format(time, yyyy-MM-dd HH:00:00) )然后在 Prophet 读取时再明确指定解析格式df_model pd.read_csv(air_feature_hourly.csv, parse_dates[ds])这看起来只是一个小细节但如果你没做这个转换等到建模阶段再排查会浪费大量时间。数据从 Spark 到 pandas 之间的“时区漂移”和“日期格式漂移”我会优先在源头解决绝不在下游临时补。6.2 缺失值处理与预测置信区间膨胀模型对连续缺失超过 3 小时的数据在缺失时段之后往往会给出特别宽的置信区间。这一点业务上要有预期。我的做法是模型输出结果时同时输出当天的数据完整度缺失小时数/24如果数据完整度低于 70%就在可视化页面上打一个“该日预测可靠性低”的标签。这个设计在真实运行里救了很多次。某几次因为上游采集设备断电当天数据完整度不到 50%模型虽然也能给出预测但区间宽度明显异常明显不可信。如果没有数据完整度标签业务方直接把预测值拿去发通报大概率要出问题。6.3 定时调度与模型落盘预测系统不能每天人工跑我最后的做法是每天 06:00 由调度器触发 PySpark 清洗作业。清洗完成后触发 Prophet 训练脚本复用历史数据 更新到昨天 23:00。训练完成后自动生成未来 7 天预测并输出到结果表。可视化服务每天 07:00 刷新页面数据。调度这块我用的是简单的crontab 日志记录。每次执行都往日志文件里追加三行作业开始时间、关键步骤完成状态、预测结果摘要比如“预测 7 天均值 68 μg/m³”。这样做的好处是第二天早上扫一眼日志就能判断昨晚的作业是否正常不需要去看一堆运行状态页面。6.4 模型重训练频率的取舍Prophet 模型并不需要每次预测都重新全量训练。如果历史数据每天只增加一条模型参数变化极小重训性价比低。我的做法是每周日全量重训一次平时每天只用最新数据做预测。代码上通过判断“今天是周几”来决定是调用model.fit()还是直接加载上次的模型文件。import pickle from datetime import datetime model_path /models/air_prophet_v1.pkl if datetime.today().weekday() 6: model train_prophet(df_all) with open(model_path, wb) as f: pickle.dump(model, f) else: with open(model_path, rb) as f: model pickle.load(f) # 每天执行预测 forecast predict_next_7d(model, future_weather)在真实生产里这个“周日重训、每日预测”的节奏比天天重训要稳定得多也节约了大量无效计算时间。最后再分享一个小经验如果你也准备在 PySpark 清洗层和 Prophet 建模层之间传数据建议从一开始就把“日期字符串标准化”和“缺失率监控”这两个问题纳入设计而不是等跑出问题再补救。这两件事在架构图上不起眼却恰恰是这类混合系统在实际运行中维护成本最高的地方。整体方案跑顺后这套“Hadoop 管数据、Prophet 管预测、可视化管表达”的组合在类似的环保监测项目中是可以直接复用的换一个城市、换一种污染物只要清洗逻辑跟着调整就行。
阅读完成 · 觉得有帮助？
