如何在PySpark中使用statsmodels的seasonal_decompose及转换时序数组

核心思路：利用Pandas中转处理 #

seasonal_decompose是statsmodels为Pandas设计的时序分解工具，直接操作PySpark数据结构会报错，正确做法是将PySpark DataFrame转换为带时间索引的Pandas DataFrame后再调用函数。

具体步骤 #

1. PySpark转Pandas并处理时间索引
   先将PySpark的time_key转为日期类型，再转成Pandas DataFrame，最后把time_key转为datetime类型并设为索引（确保时序有序）：

from pyspark.sql.functions import to_date
import pandas as pd
from statsmodels.tsa.seasonal import seasonal_decompose

# 将PySpark DataFrame转换为Pandas，同时转换日期列
pandas_df = spark_df.select("time_key", "qty") \
    .withColumn("time_key", to_date(col("time_key"))) \
    .toPandas()

# 转换为datetime类型并设置为索引，同时排序
pandas_df["time_key"] = pd.to_datetime(pandas_df["time_key"])
pandas_df.set_index("time_key", inplace=True)
pandas_df = pandas_df.sort_index()

2. 调用seasonal_decompose
   由于你的数据是7天间隔，周期可以设为52（对应一年的周数，适合年季节性分解）：

# 加法模型（根据数据特性也可以用 multiplicative）
decompose_result = seasonal_decompose(pandas_df["qty"], model="additive", period=52)

# 可查看分解后的各部分：趋势、季节性、残差
decompose_result.trend
decompose_result.seasonal
decompose_result.resid

你之前报错的原因 #

你直接将数据转为数组传递给seasonal_decompose，但该函数需要的是带时间索引的Pandas Series/DataFrame，而非单纯的键值对数组。数组中的日期会被函数误判为数值输入，因此触发类型转换错误。

PySpark中的替代方案 #

如果数据量极大无法转Pandas，可以用PySpark窗口函数实现类似的时序分解逻辑：

from pyspark.sql import Window
from pyspark.sql.functions import avg, col, row_number

# 1. 计算趋势项：用滑动窗口平均（这里用52期中心窗口）
trend_window = Window.orderBy("time_key").rowsBetween(-26, 26)
df_with_trend = spark_df.withColumn("trend", avg(col("qty")).over(trend_window))

# 2. 计算季节性：先标记每个时间点在周期内的位置（按52个周期单位）
cycle_window = Window.orderBy("time_key")
df_with_cycle = df_with_trend.withColumn("cycle_pos", (row_number().over(cycle_window) - 1) % 52 + 1)

# 计算每个周期位置的平均季节性偏移
seasonal_avg = df_with_cycle.groupBy("cycle_pos").agg(avg(col("qty") - col("trend")).alias("seasonal"))

# 合并得到分解后的完整数据
df_decomposed = df_with_cycle.join(seasonal_avg, on="cycle_pos", how="left") \
    .withColumn("residual", col("qty") - col("trend") - col("seasonal"))

这种方式的效果和seasonal_decompose接近，但需要手动调整窗口大小和周期参数来匹配你的数据特性。

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