1. 项目背景与目标
去年接手公司销售预测任务时，我面临一个典型的数据科学难题：如何利用历史销售数据构建可靠的预测模型。传统Excel表格和简单线性回归已经无法满足业务需求，我们需要处理的是TB级的交易记录、数百个SKU以及复杂的季节性因素。
经过技术选型，最终选择了Spark ML作为解决方案。Spark的分布式计算能力可以轻松处理海量数据，而MLlib提供的丰富算法库则覆盖了从特征工程到模型训练的全流程。这个决策背后有几个关键考量：
- 数据规模：单机Python+Pandas在百万级数据上尚可运行，但当数据量达到千万级时，内存和计算时间都成为瓶颈
- 特征复杂度：需要同时处理数值型特征（如价格）、类别型特征（如地区）和时间序列特征
- 生产环境要求：模型需要定期（每日）重新训练，并能够无缝集成到现有Java技术栈中
提示：Spark特别适合需要迭代计算的机器学习任务，因为其内存计算特性可以避免反复读写磁盘带来的性能损耗
2. 环境搭建与数据准备
2.1 Spark集群部署方案
在实际部署中，我选择了Standalone模式而非YARN或Kubernetes，主要基于以下考虑：
开发测试环境：本地模式（Local Mode）
# 下载Spark 3.3.2预编译版本 wget https://archive.apache.org/dist/spark/spark-3.3.2/spark-3.3.2-bin-hadoop3.tgz tar -xzf spark-3.3.2-bin-hadoop3.tgz cd spark-3.3.2-bin-hadoop3 # 启动PySpark shell ./bin/pyspark --master local[4]生产环境：3节点集群（1 Master + 2 Worker）
- Master节点：16核CPU，64GB内存
- Worker节点：各8核CPU，32GB内存
- 配置参数示例：
spark.executor.memory=24G spark.driver.memory=8G spark.executor.cores=4
2.2 数据源接入与清洗
原始销售数据存储在MySQL数据库中，包含以下关键表：
| 表名 | 数据量 | 主要字段 |
|---|---|---|
| sales_order | 1200万行 | order_id, product_id, quantity, price, order_date |
| products | 850行 | product_id, category, price, cost |
| customers | 3.2万行 | customer_id, region, industry |
数据清洗的关键步骤：
- 处理缺失值：对数值型字段用中位数填充，类别型字段用"UNKNOWN"标记
- 异常值检测：使用IQR方法识别并修正异常销售额
from pyspark.sql.functions import col, when q1, q3 = df.approxQuantile("amount", [0.25, 0.75], 0.05) iqr = q3 - q1 df_clean = df.withColumn("amount", when(col("amount") > q3 + 1.5*iqr, q3 + 1.5*iqr) .when(col("amount") < q1 - 1.5*iqr, q1 - 1.5*iqr) .otherwise(col("amount"))) - 时间特征工程：提取年、月、周、日等时间维度
3. 特征工程实战
3.1 构建特征管道
Spark ML的Pipeline API极大简化了特征处理流程。以下是核心特征转换步骤：
from pyspark.ml.feature import ( VectorAssembler, StringIndexer, OneHotEncoder, MinMaxScaler ) # 类别型特征处理 category_cols = ["product_category", "region"] indexers = [StringIndexer(inputCol=c, outputCol=f"{c}_index") for c in category_cols] encoders = [OneHotEncoder(inputCol=f"{c}_index", outputCol=f"{c}_encoded") for c in category_cols] # 数值型特征归一化 numeric_cols = ["price", "historical_avg"] assembler_numeric = VectorAssembler(inputCols=numeric_cols, outputCol="numeric_features") scaler = MinMaxScaler(inputCol="numeric_features", outputCol="scaled_features") # 时间特征处理 time_cols = ["day_of_week", "month"] assembler_time = VectorAssembler(inputCols=time_cols, outputCol="time_features") # 最终特征合并 final_assembler = VectorAssembler( inputCols=["product_category_encoded", "region_encoded", "scaled_features", "time_features"], outputCol="features" )3.2 特征重要性分析
在模型训练前，我使用随机森林进行了特征重要性评估：
| 特征 | 重要性得分 |
|---|---|
| 历史同期销售额 | 0.32 |
| 产品类别 | 0.18 |
| 促销活动 | 0.15 |
| 季节性指数 | 0.12 |
| 地区经济指标 | 0.08 |
| 节假日标志 | 0.07 |
| 天气数据 | 0.05 |
| 竞品价格 | 0.03 |
这个分析帮助我们剔除了重要性低于0.05的特征，减少了模型复杂度。
4. 模型训练与优化
4.1 算法选型对比
测试了三种主流算法并比较其表现：
| 算法 | RMSE | 训练时间 | 内存消耗 | 适用性 |
|---|---|---|---|---|
| 随机森林 | 1245 | 38min | 高 | 非线性关系 |
| GBT | 1187 | 52min | 中 | 时序特征 |
| 线性回归 | 1562 | 12min | 低 | 基线模型 |
最终选择梯度提升树(GBT)作为基础算法，因为：
- 对时序数据表现优异
- 可以自动处理特征间的交互作用
- 提供特征重要性评估
4.2 超参数调优
使用CrossValidator进行网格搜索：
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator paramGrid = (ParamGridBuilder() .addGrid(gbt.maxDepth, [5, 10]) .addGrid(gbt.maxIter, [50, 100]) .addGrid(gbt.stepSize, [0.01, 0.1]) .build()) evaluator = RegressionEvaluator( metricName="rmse", labelCol="sales_amount", predictionCol="prediction") cv = CrossValidator( estimator=gbt, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=3) cv_model = cv.fit(train_data)最佳参数组合：
- maxDepth: 10
- maxIter: 100
- stepSize: 0.1
4.3 模型评估指标
在测试集上的表现：
| 指标 | 值 | 业务含义 |
|---|---|---|
| RMSE | 1124 | 平均误差约1124元 |
| MAE | 876 | 50%预测误差在876元内 |
| R² | 0.83 | 模型解释83%的方差 |
5. 生产部署与监控
5.1 模型持久化与加载
# 保存模型 cv_model.write().overwrite().save("hdfs://path/to/sales_prediction_model") # 加载模型 from pyspark.ml.regression import GBTRegressionModel model = GBTRegressionModel.load("hdfs://path/to/sales_prediction_model")5.2 批处理预测流程
设计为每日凌晨运行的Spark作业：
- 从数据仓库加载最新销售数据
- 执行相同的特征工程流程
- 调用模型进行预测
- 将结果写入MySQL结果表
- 发送预测报告邮件
调度系统配置示例：
# 使用Airflow调度 spark-submit \ --master yarn \ --deploy-mode cluster \ sales_prediction.py \ --date $(date +%Y-%m-%d)5.3 模型监控与迭代
建立的关键监控指标：
| 指标 | 阈值 | 检查频率 |
|---|---|---|
| 预测准确率 | ±15% | 每日 |
| 特征缺失率 | <5% | 每周 |
| 模型漂移检测 | KS<0.2 | 每月 |
| 训练时间 | <1h | 每次训练 |
当监控到性能下降时触发重新训练流程：
- 收集新增数据
- 验证数据质量
- 增量训练或全量重新训练
- A/B测试新旧模型
- 生产环境切换
6. 踩坑经验与优化技巧
6.1 内存优化实战
遇到过的OOM问题及解决方案：
问题：执行collect()时Driver内存溢出
- 解决：改用take()或limit()获取样本数据
问题：特征工程阶段Executor频繁挂掉
- 解决：调整分区数
df = df.repartition(200) # 确保每个分区约100MB数据问题：模型评估时内存不足
- 解决：使用近似评估方法
evaluator = RegressionEvaluator( metricName="rmse", labelCol="label", predictionCol="prediction") # 对1%样本进行评估 sample_test = test_data.sample(0.01) evaluator.evaluate(sample_test)
6.2 性能调优参数
经过多次测试得出的最佳配置：
spark.conf.set("spark.sql.shuffle.partitions", "200") spark.conf.set("spark.executor.memoryOverhead", "2g") spark.conf.set("spark.dynamicAllocation.enabled", "true") spark.conf.set("spark.shuffle.service.enabled", "true")6.3 业务经验分享
- 节假日处理：为特殊日期（如双11）创建单独的特征标志
- 新产品预测：对于没有历史数据的新品，采用同类产品均值法
- 预测结果解释：为业务部门提供预测区间而非单点估计
from pyspark.sql.functions import expr predictions = predictions.withColumn( "prediction_interval", expr("concat(round(prediction*0.9), '-', round(prediction*1.1))") )
7. 扩展应用与未来改进
当前系统已经稳定运行8个月，平均预测准确率达到87%。接下来计划从三个方向进行优化：
实时预测：探索Structured Streaming实现近实时预测
from pyspark.sql.streaming import DataStreamReader stream = (spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "host1:port1,host2:port2") .option("subscribe", "sales_topic") .load()) # 实时特征工程和预测 predictions = model.transform(stream)集成外部数据：引入天气、经济指标等外部数据源
模型融合：尝试将时间序列模型（如Prophet）与机器学习模型结合
在实施过程中，最大的体会是：Spark ML虽然强大，但必须根据业务特点进行充分定制。单纯追求算法复杂度往往适得其反，好的预测系统需要在数据质量、特征工程和业务理解之间找到平衡点。
