PySpark中Prophet时序模型报错：DataFrame非NaN行不足2行求助

错误信息 #

PythonException: 'ValueError: Dataframe has less than 2 non-NaN rows.',
from , line 17. Full traceback below:
org.apache.spark.SparkException: Job aborted due to stage failure:
Task 0 in stage 367.0 failed 4 times, most recent failure: Lost task
0.3 in stage 367.0 (TID 541) (172.26.145.6 executor 0): org.apache.spark.api.python.PythonException: 'ValueError: Dataframe
has less than 2 non-NaN rows.', from , line
17. Full traceback below: Traceback (most recent call last): File "", line 17, in pd_apply_forecast File
"/databricks/python/lib/python3.8/site-packages/prophet/forecaster.py",
line 1113, in fit
raise ValueError('Dataframe has less than 2 non-NaN rows.') ValueError: Dataframe has less than 2 non-NaN rows.

问题定位 #

报错核心是Prophet模型拟合时，按groupby_col分组后的单组数据非NaN行数不足2条，无法满足时间序列建模的最小数据量要求。问题出在两个环节：

1. 上游分组逻辑未过滤数据量不足的分组，导致无效分组进入拟合流程；
2. pd_apply_forecast函数未提前校验数据量，直接调用model.fit()触发报错。

修复方案及代码修改位置 #

1. 提前过滤无效分组（修改df_bizlist生成逻辑） #

在生成业务分组列表时，直接过滤掉数据量少于2的分组，从源头避免无效数据进入后续流程：

df_bizlist = (
    df.filter(f"{date_col} >= coalesce(date_sub(date 'today', {num_days_check}), '{start_date}')")
        .groupBy(groupby_col)
        .count()
        .filter(col("count") >= 2)  # 新增：过滤数据量不足2的分组
        .orderBy(col("count").desc())
)

2. 在拟合函数中增加数据量校验（修改pd_apply_forecast函数） #

在调用model.fit()前检查数据行数，不足2行时直接返回符合输出schema的空DataFrame，避免报错：

def pd_apply_forecast(pd_history: pd.DataFrame) -> pd.DataFrame: 
    # 过滤空业务段及关键列缺失值，明确指定校验列避免误删有效数据
    pd_history = (pd_history[pd_history[groupby_col].notnull()]
                      .dropna(subset=['ds', 'y']))
    
    # 新增：数据量校验，不足2行直接返回空结果
    if len(pd_history) < 2:
        empty_cols = [groupby_col, 'ds', 'y', 'yhat', 'yhat_lower', 'yhat_upper', 'trend', 'trend_lower', 'trend_upper', 'additive_terms', 'additive_terms_lower', 'additive_terms_upper', 'weekly', 'weekly_lower', 'weekly_upper', 'yearly', 'yearly_lower', 'yearly_upper', 'multiplicative_terms', 'multiplicative_terms_lower', 'multiplicative_terms_upper']
        return pd.DataFrame(columns=empty_cols)
    
    # 模型初始化及拟合逻辑不变
    model = Prophet(
        growth='linear',
        yearly_seasonality='auto',
        weekly_seasonality='auto',
        daily_seasonality=False,
        seasonality_mode='additive'
    )
    
    model.fit(pd_history)
    
    # 后续预测结果拼接逻辑不变...

3. 变量名一致性修正 #

注意函数参数groupby_cols与代码内部使用的groupby_col是否匹配：如果groupby_cols是多列列表，需统一变量名；如果是单列，建议将函数参数改为groupby_col，避免因变量名不一致导致数据丢失。

完整修改后代码 #

import re
import pandas as pd
import pyspark.pandas as ps
from prophet import Prophet
from pyspark.sql import DataFrame, StructType, StructField, StringType, DateType, IntegerType, FloatType
from pyspark.sql.functions import col, lit, coalesce, expr, array_max
from datetime import date, timedelta

def run_row_outlier_check(df: DataFrame, min_date, start_date, groupby_col, job_id) -> DataFrame:
    """
    Generate dataframe containing prophet model forecasting of row counts
    """
    pd_schema = StructType([
        StructField(groupby_col, StringType(), True),
        StructField("ds", DateType(), True),
        StructField("y", IntegerType(), True),
        StructField("yhat", FloatType(), True),
        StructField("yhat_lower", FloatType(), True),
        StructField("yhat_upper", FloatType(), True),
        StructField("trend", FloatType(), True),
        StructField("trend_lower", FloatType(), True),
        StructField("trend_upper", FloatType(), True),
        StructField("additive_terms", FloatType(), True),
        StructField("additive_terms_lower", FloatType(), True),
        StructField("additive_terms_upper", FloatType(), True),
        StructField("weekly", FloatType(), True),
        StructField("weekly_lower", FloatType(), True),
        StructField("weekly_upper", FloatType(), True),
        StructField("yearly", FloatType(), True),
        StructField("yearly_lower", FloatType(), True),
        StructField("yearly_upper", FloatType(), True),
        StructField("multiplicative_terms", FloatType(), True),
        StructField("multiplicative_terms_lower", FloatType(), True),
        StructField("multiplicative_terms_upper", FloatType(), True)
        ])

    # 生成连续日期DataFrame
    df_rundates = (ps.DataFrame({'date':pd.date_range(start=min_date, end=(date.today() - timedelta(days=1)))})).to_spark()

    # 生成业务分组列表并过滤数据量不足的组
    df_bizlist = (
        df.filter(f"{date_col} >= coalesce(date_sub(date 'today', {num_days_check}), '{start_date}')")
            .groupBy(groupby_col)
            .count()
            .filter(col("count") >= 2)
            .orderBy(col("count").desc())
    )

    df_rundates_bus = (
        df_rundates
            .join(df_bizlist, how='full')
            .select(df_bizlist[groupby_col], df_rundates["date"].alias("ds"))
    )

    # 构建Prophet输入DataFrame
    df_grouped_cnt = df.groupBy(groupby_col, date_col).count()
    df_input = (
        df_rundates_bus.selectExpr(f"{groupby_col}", "to_date(ds) as ds")
            .join(df_grouped_cnt.selectExpr(f"{groupby_col}", f"{date_col} as ds", "count as y"), on=['ds',f'{groupby_col}'], how='left')
            .withColumn("y", coalesce("y", lit(0)))
            .repartition(sc.defaultParallelism, "ds")
    )
    
    # 执行预测
    df_forecast = (
        df_input
            .groupBy(groupby_col)
                .applyInPandas(pd_apply_forecast, schema=pd_schema)
    )
    
    # 生成异常值结果
    df_rowoutliers = (
        df_forecast
            .filter("y > 0 AND (y > yhat_upper OR y < array_max(array(yhat_lower,0)))")
            .withColumn("check_type", lit("row_count"))
            .withColumn("deduct_score", expr("round(sqrt(pow(y-yhat, 2) / pow(yhat_lower - yhat_upper,2)))").cast('int'))
            .select(
                col("check_type"),
                col("ds").alias("ref_date"),
                col(groupby_col).alias("ref_dimension"),
                col("y").cast('int').alias("actual"),
                col("deduct_score"),
                col("yhat").alias("forecast"),
                col("yhat_lower").alias("forecast_lower"),
                col("yhat_upper").alias("forecast_upper")
            )
    )
    
    return add_metadata_columns(df_forecast, job_id), add_metadata_columns(df_rowoutliers, job_id)

def pd_apply_forecast(pd_history: pd.DataFrame) -> pd.DataFrame: 
    # 过滤空业务段及关键列缺失值
    pd_history = (pd_history[pd_history[groupby_col].notnull()]
                      .dropna(subset=['ds', 'y']))
    
    # 数据量校验，不足2行返回空结果
    if len(pd_history) < 2:
        empty_cols = [groupby_col, 'ds', 'y', 'yhat', 'yhat_lower', 'yhat_upper', 'trend', 'trend_lower', 'trend_upper', 'additive_terms', 'additive_terms_lower', 'additive_terms_upper', 'weekly', 'weekly_lower', 'weekly_upper', 'yearly', 'yearly_lower', 'yearly_upper', 'multiplicative_terms', 'multiplicative_terms_lower', 'multiplicative_terms_upper']
        return pd.DataFrame(columns=empty_cols)
    
    # 初始化模型
    model = Prophet(
        growth='linear',
        yearly_seasonality='auto',
        weekly_seasonality='auto',
        daily_seasonality=False,
        seasonality_mode='additive'
    )
    
    # 拟合模型
    model.fit(pd_history)
    
    # 生成未来日期
    pd_future = model.make_future_dataframe(
        periods=365, 
        freq='d',
        include_history=True
    )
    
    # 预测
    pd_forecast = model.predict(pd_future)
    
    # 拼接历史数据与预测结果
    pd_f = pd_forecast[ ['ds', 'yhat', 'yhat_lower', 'yhat_upper', 'trend', 'trend_lower', 'trend_upper', 'additive_terms', 'additive_terms_lower', 'additive_terms_upper', 'weekly', 'weekly_lower', 'weekly_upper', 'yearly', 'yearly_lower', 'yearly_upper', 'multiplicative_terms', 'multiplicative_terms_lower', 'multiplicative_terms_upper'] ].set_index('ds')

    pd_h = pd_history[ ['ds', groupby_col, 'y'] ].set_index('ds')

    pd_results = pd_f.join(pd_h, how='left')
    pd_results.reset_index(level=0, inplace=True)
    
    # 过滤空业务段
    pd_results = pd_results[pd_results[groupby_col].notnull()]
    
    # 返回结果
    return pd_results[ [groupby_col, 'ds', 'y', 'yhat', 'yhat_lower', 'yhat_upper', 'trend', 'trend_lower', 'trend_upper', 'additive_terms', 'additive_terms_lower', 'additive_terms_upper', 'weekly', 'weekly_lower', 'weekly_upper', 'yearly', 'yearly_lower', 'yearly_upper', 'multiplicative_terms', 'multiplicative_terms_lower', 'multiplicative_terms_upper'] ]

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