使用HyperOpt的SparkTrials时如何在工作节点创建Spark DataFrame

SparkTrials的运行逻辑是将objective函数序列化后分发到Spark Executor节点并行执行，而SparkSession、SparkContext属于仅能在驱动节点创建和访问的对象，Executor环境不允许初始化或调用这类对象，这是Spark分布式架构的固有约束，和获取SparkSession的写法无关。

你不需要重写原有预处理逻辑，选择以下任一方案即可：

方案一（推荐：不损失并行度，代码改动最小） #

调整代码执行顺序，将Spark DataFrame创建、预处理逻辑移动到所有试验运行完成后，在驱动节点统一执行：

• 目标函数仅返回超参、模型指标、以及你需要用于后续预处理的原始结果
• 所有超参搜索完成后，在驱动端收集所有试验的返回结果，统一创建Spark DataFrame，直接复用原有预处理代码
  这个方案完全不需要修改原有预处理逻辑，也不会影响HyperOpt的并行试验效率。

方案二（仅小批量试验适用） #

如果试验数量少，不需要太高的并行度，可以直接放弃SparkTrials，使用普通Trials运行，所有目标函数都在驱动节点执行，自然可以正常创建Spark DataFrame。

from sklearn.datasets import load_iris
from sklearn.model_selection import cross_val_score
from sklearn.svm import SVC
from hyperopt import fmin, tpe, hp, SparkTrials, STATUS_OK, Trials
from pyspark.sql import SparkSession
import mlflow

# 加载数据集
iris = load_iris()
X = iris.data
y = iris.target

def objective(C):
    clf = SVC(C)
    # 目标函数内仅做模型训练、评估，返回需要的所有原始数据
    accuracy = cross_val_score(clf, X, y).mean()
    # 把你需要转Spark DF的结果随返回值一起带回，这里示例返回你之前的测试数据
    raw_data = [('Alice', 1)]
    return {'loss': -accuracy, 'status': STATUS_OK, 'raw_data': raw_data, 'C': C}

search_space = hp.lognormal('C', 0, 1.0)
algo=tpe.suggest

# 使用SparkTrials并行运行试验
spark_trials = SparkTrials(parallelism=4)
argmin = fmin(
  fn=objective,
  space=search_space,
  algo=algo,
  max_evals=16,
  trials=spark_trials)

# 所有试验完成后，在驱动端统一收集结果、创建Spark DataFrame
ss = SparkSession.builder.getOrCreate()
all_result = []
for trial in spark_trials.results:
    if trial['status'] == STATUS_OK:
        # 组装你需要的字段，这里直接取目标函数返回的raw_data
        all_result.extend(trial['raw_data'])
# 驱动端创建Spark DF，直接调用原有预处理逻辑即可
sdf = ss.createDataFrame(all_result, schema=['name', 'id'])

# 这里直接复用你原来的预处理代码即可，无需任何修改

• 不要尝试在Executor端调用任何Spark驱动端专属API，包括创建DataFrame、读写Spark表等操作，这类操作即使临时跑通也会有资源泄漏、任务不稳定的问题。
• 如果原有逻辑要求每个试验单独处理，你可以在驱动端遍历每个trial的结果，逐个转Spark DF再调用预处理逻辑，效果和你原来在目标函数里处理完全一致。

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