基于Spark ML 3.1+的销量时间序列预测方案咨询

你之前的线性回归实现存在两处核心错误：

• 特征列错误引入了label（销量）字段，相当于让模型直接用真实销量拟合，得到的R²=1属于无效的完美拟合结果，完全不具备预测未来的能力
• 时间戳作为连续值直接输入线性回归，无法捕捉时间序列的趋势、周期、季节效应等规律，不适合单变量时序预测场景

推荐使用滑动窗口构造时序滞后特征+梯度提升树(GBT)回归方案，该方案全版本兼容Spark ML 3.1+，无额外依赖，适配超大规模数据集训练，预测精度优于普通线性回归。

import org.apache.spark.ml.feature.VectorAssembler;
import org.apache.spark.ml.regression.GBTRegressor;
import org.apache.spark.ml.regression.GBTRegressionModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.IntegerType;
import org.apache.spark.sql.types.LongType;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.RowFactory;
import java.util.Arrays;
import static org.apache.spark.sql.functions.*;

// 1. 数据预处理：构造时序特征
Dataset<Row> processedData = dataSetBanana
        // 时序数据必须按时间升序排序
        .orderBy(col("time").asc())
        // 提取时间衍生特征，捕捉周/月周期规律
        .withColumn("year", year(to_timestamp(col("time"))))
        .withColumn("month", month(to_timestamp(col("time"))))
        .withColumn("day_of_month", dayofmonth(to_timestamp(col("time"))))
        .withColumn("day_of_week", dayofweek(to_timestamp(col("time"))))
        // 构造过去7天的销量滞后特征，窗口大小可根据数据规律调整为14/30天
        .withColumn("lag1", lag(col("label"), 1).over(Window.orderBy("time")))
        .withColumn("lag2", lag(col("label"), 2).over(Window.orderBy("time")))
        .withColumn("lag3", lag(col("label"), 3).over(Window.orderBy("time")))
        .withColumn("lag4", lag(col("label"), 4).over(Window.orderBy("time")))
        .withColumn("lag5", lag(col("label"), 5).over(Window.orderBy("time")))
        .withColumn("lag6", lag(col("label"), 6).over(Window.orderBy("time")))
        .withColumn("lag7", lag(col("label"), 7).over(Window.orderBy("time")))
        // 剔除前7条无滞后特征的脏数据
        .na().drop();

// 2. 组装特征向量
VectorAssembler assembler = new VectorAssembler()
        .setInputCols(new String[]{"year", "month", "day_of_month", "day_of_week", "lag1", "lag2", "lag3", "lag4", "lag5", "lag6", "lag7"})
        .setOutputCol("features");
Dataset<Row> featureData = assembler.transform(processedData);

// 3. 时序数据集切分：按时间顺序切前80%为训练集，后20%为验证集，禁止随机切分
long totalCnt = featureData.count();
Dataset<Row> trainData = featureData.limit((long)(totalCnt * 0.8));
Dataset<Row> valData = featureData.except(trainData);

// 4. 训练GBT回归模型
GBTRegressor gbt = new GBTRegressor()
        .setLabelCol("label")
        .setFeaturesCol("features")
        .setMaxIter(50)
        .setMaxDepth(5);
GBTRegressionModel gbtModel = gbt.fit(trainData);

// 5. 预测未来指定日期销量
// 构造未来日期的特征：需填充对应时间衍生特征和历史7天销量
Dataset<Row> futureInput = spark.createDataFrame(
        Arrays.asList(RowFactory.create(1626414043, 2021, 7, 16, 5, 44, 42, 41, 43, 42, 42, 41)),
        new StructType()
                .add("time", LongType)
                .add("year", IntegerType)
                .add("month", IntegerType)
                .add("day_of_month", IntegerType)
                .add("day_of_week", IntegerType)
                .add("lag1", IntegerType)
                .add("lag2", IntegerType)
                .add("lag3", IntegerType)
                .add("lag4", IntegerType)
                .add("lag5", IntegerType)
                .add("lag6", IntegerType)
                .add("lag7", IntegerType)
);
Dataset<Row> futureFeature = assembler.transform(futureInput);
Dataset<Row> predictionResult = gbtModel.transform(futureFeature)
        .select(col("time"), col("prediction").cast(IntegerType).alias("predicted_label"));

// 输出结果
predictionResult.show(false);

+----------+---------------+
|time      |predicted_label|
+----------+---------------+
|1626414043|38             |
+----------+---------------+

内容的提问来源于stack exchange，提问作者PatPanda