使用Pandas UDF并行化Prophet模型超参数调优遇性能瓶颈求助

问题描述 #

我尝试对Prophet模型的超参数进行并行调优，共有约100组超参数存储在DataFrame params_df中。为实现并行调优，编写了如下代码：

调优UDF定义 #

schema = StructType([StructField('hyperparameters', StringType(), True),
                 StructField('mape', DoubleType(), True)]) 

@pandas_udf(schema,PandasUDFType.GROUPED_MAP)
def grid_search(hyperparameters):
    train=valid_train_df.drop(columns='Series').copy()
    cutoffs=[train.loc[len(train)-5,'ds'],train.loc[len(train)-3,'ds']]
    hyperparameters=hyperparameters.reset_index(drop=True)
    params=ast.literal_eval(hyperparameters['hyperparameters'][0])

    prophet_test = Prophet(**params)

    for regressor in indep:
        prophet_test.add_regressor(regressor)

    prophet_test.fit(train[indep+['ds','y']])

    df_cv = cross_validation(model=prophet_test, horizon='62 days', cutoffs=cutoffs)
    df_cv['monthly_mape']=abs(df_cv['y']-df_cv['yhat'])/df_cv['y']
    mape=df_cv['monthly_mape'].mean()
    
    tuning_results = pd.DataFrame({'hyperparameters':str(params),'mape':mape},index=[0])

    return tuning_results

并行执行代码 #

run_udf_results=spark_parallel_df.groupby('hyperparameters').apply(grid_search)
hyperparameter_results=run_udf_results.toPandas()

其中valid_train_df为输入数据集，spark_parallel_df是params_df对应的Spark DataFrame。但实际运行耗时约40分钟，未达到预期的并行性能，想请教是否存在疏漏？

核心问题与优化方案 #

1. 分组方式导致任务调度开销过大 #

你用groupby('hyperparameters').apply(grid_search)，但如果spark_parallel_df里每个hyperparameters对应唯一一行数据（100组超参数对应100行），分组后每个组仅含1条数据。Spark的GROUPED_MAP针对批量分组优化，单条数据分组会产生大量调度任务，开销远超实际计算成本。

优化方案：
改用PandasUDFType.SCALAR类型的UDF，或者直接用mapPartitions让每个分区处理一批超参数，减少调度次数。也可以将超参数按固定批次打包成分组，降低任务数量。

2. 全局训练数据的序列化与传输冗余 #

UDF中直接引用全局变量valid_train_df，Spark会把整个DataFrame序列化后传递给每个Executor，既增加网络传输负担，又可能引发序列化性能问题。此外每次UDF执行时的copy()操作完全多余。

优化方案：
将训练数据通过Spark广播变量传递，避免重复序列化和传输：

# 在UDF定义前广播训练数据
broadcast_train = spark.sparkContext.broadcast(valid_train_df.drop(columns='Series'))

@pandas_udf(schema, PandasUDFType.GROUPED_MAP)
def grid_search(hyperparameters):
    train = broadcast_train.value  # 从广播变量获取训练数据
    # 后续逻辑保持不变

3. Cross Validation的计算量冗余 #

你设置了2个cutoffs，每组超参数需要执行2次模型训练+预测，100组超参数对应200次完整模型流程，这是硬计算开销。

优化方案：

• 评估horizon='62 days'是否必要，可考虑缩短验证窗口减少计算量
• 若业务允许，减少cutoffs数量甚至改用单次验证替代交叉验证

4. Spark集群资源配置不足 #

如果集群分配的Executor数量过少、每个Executor的CPU/内存不足，100组超参数无法真正并行执行，只能串行或少量并行。

优化方案：
调整Spark配置：

• 增加spark.executor.instances，设置为与集群CPU核心数匹配的值
• 调整spark.executor.cores，每个Executor分配2-4核（根据集群资源调整）
• 确保spark.driver.memory和spark.executor.memory足够容纳训练数据与模型

5. UDF内部冗余操作优化 #

• 分组后每组仅一行数据，可直接用hyperparameters.iloc[0]['hyperparameters']替代reset_index+取索引0的操作
• 若spark_parallel_df中的hyperparameters字段已定义为Spark的MapType，可直接读取结构化数据，避免ast.literal_eval的字符串解析开销

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