如何加载Spark已保存Pipeline并结合新数据实现增量重训练？

问题背景 #

需要加载Spark中已保存的Pipeline，采用每日新增数据的策略重新训练模型，但加载环节出现两类错误：

• 使用PipelineModel.load()加载时，出现“no fit() method”错误
• 使用Pipeline().load()加载时，报错“Error loading metadata: Expected class name org.apache.spark.ml.Pipeline but found class name org.apache.spark.ml.PipelineModel”

原代码：

new_data_df = data in current day
if target path exists:
  model = PipelineModel.load("path/to/pipeline")
  first_round = model.transform(new_data_df)
  evaluator = BinaryClassificationEvaluator()
  evaluator.evaluate(first_round)
else:
  assembler = VectorAssembler().setInputCols(ft_cols).setOutputCol('features')
  lr = LogisticRegression(maxIter=150, elasticNetParam=0.3, regParam=0.01, labelCol=target, featuresCol='features',
                                standardization=False,
                                predictionCol='prediction')
  model = Pipeline().setStages([assembler, lr])

trained_model = model.fit(new_data_df)

lrm = trained_model.stages[-1]
trainingSummary = lrm.summary
objectiveHistory = trainingSummary.objectiveHistory
trained_model.save("path/to/model/current date")

错误原因 #

• PipelineModel是训练完成的模型实例，仅包含预测（transform）能力，没有fit()方法，无法用于重新训练。
• 保存的对象是PipelineModel（训练后的模型），而Pipeline().load()只能加载未训练的Pipeline工作流，类型不匹配导致报错。

解决方案 #

利用Spark LogisticRegression的warmStart特性实现增量训练，核心思路是复用已训练模型的参数，基于新数据继续训练，同时保留原特征工程逻辑：

修改后的代码 #

from pyspark.ml import Pipeline, PipelineModel
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator
import os

new_data_df = 每日新增数据  # 替换为实际的当日数据加载逻辑
target = "你的标签列名"
ft_cols = ["特征列1", "特征列2"]  # 替换为实际特征列
model_save_path = "path/to/saved_pipeline_model"
current_date = "20240520"  # 替换为实际日期生成逻辑

if os.path.exists(model_save_path):
    # 加载已训练的PipelineModel
    existing_pipeline_model = PipelineModel.load(model_save_path)
    
    # 评估现有模型在新数据上的表现
    pred_df = existing_pipeline_model.transform(new_data_df)
    evaluator = BinaryClassificationEvaluator(labelCol=target)
    eval_result = evaluator.evaluate(pred_df)
    print(f"现有模型在当日数据上的评估分数: {eval_result}")
    
    # 提取已训练的逻辑回归模型，开启warm start
    trained_lr_model = existing_pipeline_model.stages[-1]
    lr = LogisticRegression(
        maxIter=150,
        elasticNetParam=trained_lr_model.getElasticNetParam(),
        regParam=trained_lr_model.getRegParam(),
        labelCol=target,
        featuresCol='features',
        standardization=False,
        predictionCol='prediction',
        warmStart=True,  # 开启热启动，复用已有参数
        initialModel=trained_lr_model  # 设置初始模型
    )
    
    # 复用原特征工程阶段，构建新的Pipeline
    assembler = existing_pipeline_model.stages[0]
    pipeline = Pipeline().setStages([assembler, lr])
    
    # 用当日增量数据训练
    updated_pipeline_model = pipeline.fit(new_data_df)
else:
    # 首次训练流程
    assembler = VectorAssembler().setInputCols(ft_cols).setOutputCol('features')
    lr = LogisticRegression(
        maxIter=150,
        elasticNetParam=0.3,
        regParam=0.01,
        labelCol=target,
        featuresCol='features',
        standardization=False,
        predictionCol='prediction'
    )
    pipeline = Pipeline().setStages([assembler, lr])
    updated_pipeline_model = pipeline.fit(new_data_df)

# 获取训练摘要
updated_lr_model = updated_pipeline_model.stages[-1]
training_summary = updated_lr_model.summary
objective_history = training_summary.objectiveHistory

# 保存更新后的模型（按日期区分或覆盖原路径）
updated_pipeline_model.save(f"path/to/model/{current_date}")

关键要点 #

• 复用特征工程逻辑：从已加载的PipelineModel中提取原特征处理阶段（如VectorAssembler），确保新数据的特征处理和训练时一致。
• 开启Warm Start：通过warmStart=True和initialModel参数，让LogisticRegression基于已有模型的参数继续训练，实现增量学习。
• 区分Pipeline与PipelineModel：Pipeline是未训练的工作流（可fit），PipelineModel是训练后的模型（仅可transform），保存和加载时要匹配类型。

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