PySpark ALS模型save保存报错 如何正确持久化避免重复训练

问题场景 #

在PySpark环境中使用ALS算法训练协同过滤推荐模型，核心训练代码如下：

from pyspark.ml.recommendation import ALS, ALSModel
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.mllib.evaluation import RegressionMetrics, RankingMetrics
from pyspark.ml.evaluation import RegressionEvaluator

als = ALS(maxIter=15, 
              regParam=0.08, 
              userCol="ID User", 
              itemCol="ID Film", 
              ratingCol="Rating",
              rank=20,
              numItemBlocks=30,
              numUserBlocks = 30,
              alpha = 0.95,
              nonnegative = True, 
              coldStartStrategy="drop",
             implicitPrefs=False)
model = als.fit(training_dataset)

model.save('model')

调用save方法时Jupyter Notebook抛出如下错误，模型保存失败：

An error occurred while calling o477.save.
: org.apache.spark.SparkException: Job aborted.
    at org.apache.spark.internal.io.SparkHadoopWriter$.write(SparkHadoopWriter.scala:106)

已尝试的无效方案 #

先后测试以下保存方法，均未解决问题：

• 直接调用model.save('model')
• 调用model.write().save("saved_model")
• 调用未训练的ALS实例方法als.write().save("saved_model")
• 调用als.save('model')
• 使用Python原生pickle序列化ALS对象：import pickle; s = pickle.dumps(als)
• 拼接自定义本地路径后调用als.save(als_path)，其中路径写法为als_path = "from_C:Folder_to_my_project_root" + "/als"

需求是实现训练完成的ALS模型持久化，后续程序运行时可直接加载模型，无需重复执行训练流程。

正确操作步骤 #

先明确核心认知误区 #

• als是未经过训练的算法估计器（Estimator），保存该对象没有实际业务价值，我们需要持久化的是训练完成后输出的ALSModel实例，也就是代码中的model变量。
• 禁止使用Python原生pickle序列化Spark ML/MLlib的模型对象：Spark模型的实际参数存储在JVM端的分布式对象中，pickle只能序列化Python侧的薄包装层，完全无法持久化真实模型数据，必然触发序列化错误。
• Spark的save方法默认不覆盖已存在路径，如果目标保存位置已经有同名文件/目录，会直接触发任务中止报错。

修复路径配置 #

Windows本地环境下路径写法不规范是这类报错的高频诱因：

1. 路径统一使用正斜杠/，不要用反斜杠，无需手动转义，示例：model_path = "C:/my_project/spark_als_model"
2. 确保路径对应的目录有当前用户的写入权限
3. 如果是集群环境，不要把模型存在节点本地磁盘，要存在HDFS、对象存储等所有Executor节点均可访问的共享路径，否则Executor写入时会找不到路径触发Job aborted。

增加路径预校验逻辑 #

保存前先判断目标路径是否存在，存在则先删除，避免覆盖报错：

# 初始化Hadoop文件系统操作工具
hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()
Path = spark.sparkContext._jvm.org.apache.hadoop.fs.Path
fs = Path(model_path).getFileSystem(hadoop_conf)

# 目标路径存在则递归删除
if fs.exists(Path(model_path)):
    fs.delete(Path(model_path), True)

执行保存与加载 #

完成上述配置后即可正常保存模型：

# 保存训练好的模型
model.save(model_path)

# 后续程序中直接加载模型，无需重复训练
from pyspark.ml.recommendation import ALSModel
loaded_model = ALSModel.load(model_path)

额外避坑提示 #

当前代码中使用的用户列、物品列名ID User、ID Film带有空格，部分低版本Spark在写入模型元数据时会因为列名带特殊字符触发序列化错误，建议训练前先将列名重命名为无空格、无特殊字符的格式，从根源规避这类问题：

training_dataset = training_dataset.withColumnRenamed("ID User", "user_id")\
                                   .withColumnRenamed("ID Film", "film_id")

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