XGBoost+Hyperopt+MLflow调参时Spark Python Worker崩溃问题求助

一、Spark Python Worker意外崩溃问题 #

核心可能原因 #

• Spark Worker内存资源不足（常见退出码137对应OOM）
• XGBoost、PySpark、Python版本兼容性不匹配
• 数据处理逻辑存在异常（如大表全量加载、特征列含极端值/缺失值）
• Worker进程配置不合理（如CPU核数、线程数设置过高）

解决建议 #

• 调整Spark资源配置：在SparkSession初始化时增加内存分配，避免OOM：

  spark = SparkSession.builder \
      .appName("SupplyForecast") \
      .config("spark.executor.memory", "8g") \
      .config("spark.driver.memory", "4g") \
      .config("spark.executor.cores", "4") \
      .getOrCreate()
  

• 验证版本兼容性：确保依赖版本匹配（例如XGBoost 1.6.x搭配Spark 3.2.x，Hyperopt 0.2.7搭配Python 3.8+）
• 优化数据处理：对输入数据做分区、过滤无效值，避免全量转Pandas（改用Spark XGBoost原生API减少内存占用）：

  from pyspark.ml.feature import VectorAssembler
  assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
  train_data = assembler.transform(train_data).select("features", "label")
  

• 启用Worker日志调试：配置Spark滚动日志，查看崩溃时的详细堆栈：

  spark.conf.set("spark.executor.logs.rolling.strategy", "size")
  spark.conf.set("spark.executor.logs.rolling.maxSize", "100m")
  

二、min()参数为空（无有效MLflow实验记录）问题 #

核心可能原因 #

• Hyperopt目标函数中未正确记录MLflow指标，或单次调参失败导致无有效实验记录
• MLflow实验未初始化，默认实验无数据
• 超参数搜索空间不合理（如学习率为负、树数量为0）导致模型训练失败，无指标输出

解决建议 #

• 完善MLflow记录逻辑：在目标函数中增加异常捕获，确保即使训练失败也能记录状态，避免无有效记录：

  def objective(params):
      with mlflow.start_run(nested=True):
          try:
              # 模型训练逻辑
              model = xgb.train(params, dtrain, num_boost_round=int(params["n_estimators"]))
              preds = model.predict(dtest)
              rmse = ((preds - test_data.toPandas()["label"])**2).mean()**0.5
              
              mlflow.log_params(params)
              mlflow.log_metric("rmse", rmse)
              return {"loss": rmse, "status": STATUS_OK}
          except Exception as e:
              mlflow.log_params(params)
              mlflow.log_metric("error", 1)
              mlflow.log_text(str(e), "error_log.txt")
              return {"loss": float("inf"), "status": STATUS_FAIL}
  

• 强制初始化MLflow实验：在代码开头指定实验名称，避免默认实验无数据：

  mlflow.set_experiment("仓库物资供应预测调优实验")
  

• 校验搜索空间合理性：限制超参数的有效范围，避免无效参数导致训练失败：

  space = {
      "max_depth": hp.quniform("max_depth", 3, 10, 1),
      "learning_rate": hp.uniform("learning_rate", 0.01, 0.3),  # 改用uniform避免负数值
      "n_estimators": hp.quniform("n_estimators", 100, 500, 50)
  }
  

• 安全获取最小损失：在取min前先过滤有效记录，避免空序列报错：

  runs = mlflow.search_runs()
  valid_rmse = [run["metrics.rmse"] for _, run in runs.iterrows() if not pd.isna(run["metrics.rmse"])]
  if valid_rmse:
      best_loss = min(valid_rmse)
  else:
      raise ValueError("无有效实验记录，请检查模型训练逻辑")
  

代码片段（用户提供核心逻辑） #

from hyperopt import fmin, tpe, hp, STATUS_OK
import mlflow
import mlflow.xgboost
from pyspark.sql import SparkSession
import xgboost as xgb

spark = SparkSession.builder.appName("SupplyForecast").getOrCreate()

# 加载数据
data = spark.read.parquet("path/to/data")
train_data, test_data = data.randomSplit([0.8, 0.2])

# Hyperopt搜索空间
space = {
    "max_depth": hp.quniform("max_depth", 3, 10, 1),
    "learning_rate": hp.loguniform("learning_rate", -5, 0),
    "n_estimators": hp.quniform("n_estimators", 100, 500, 50)
}

def objective(params):
    with mlflow.start_run(nested=True):
        # 转换为XGBoost DMatrix
        dtrain = xgb.DMatrix(train_data.toPandas())
        dtest = xgb.DMatrix(test_data.toPandas())
        
        model = xgb.train(params, dtrain, num_boost_round=int(params["n_estimators"]))
        preds = model.predict(dtest)
        rmse = ((preds - test_data.toPandas()["label"])**2).mean()**0.5
        
        mlflow.log_params(params)
        mlflow.log_metric("rmse", rmse)
        return {"loss": rmse, "status": STATUS_OK}

# 运行调优
best = fmin(
    fn=objective,
    space=space,
    algo=tpe.suggest,
    max_evals=20
)

# 后续报错点：min()参数为空
best_loss = min([run.data.metrics["rmse"] for run in mlflow.search_runs()])

报错日志 #

Error 1: Spark Python Worker exited unexpectedly (crashed)
Exit code: 137
Logs: ... (内存不足相关堆栈)

Error 2: ValueError: min() arg is an empty sequence
Traceback (most recent call last):
File "supply_forecast.py", line XX, in
best_loss = min([run.data.metrics["rmse"] for run in mlflow.search_runs()])
ValueError: min() arg is an empty sequence

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