能否使用已训练的Transformer修改Spark Pipeline的阶段？

问题根因说明 #

• setHandleInvalid("keep")参数本身就会自动为未知特征预留索引位，不需要手动构造假样本追加到训练集，额外添加假样本反而会污染训练数据分布，降低模型效果。
• 无法手动实例化PipelineModel、CrossValidatorModel是因为这两个类的构造方法在Spark 2.x及以上版本为包私有权限，不允许外部直接通过new关键字创建。
• 原有代码中将未训练的stringIndexerPipeline（属于Estimator类型）强转为PipelineModel（属于Transformer类型）也会触发类型转换异常。

正确实现方案 #

1. 最简方案：统一配置参数兼容未知特征 #

直接给特征处理组件配置兼容参数，不需要修改训练数据，也不需要手动拼接Pipeline：

val catIndexer = catFeatures.map(cname => {
  new StringIndexer()
    // 自动为未知特征分配最大索引值（等于训练集出现过的分类总数）
    .setHandleInvalid("keep")
    .setInputCol(cname)
    .setOutputCol(cname + KeyColumns.stringIndexerSuffix)
})
val indexedCatFeatures = catIndexer.map(idx => idx.getOutputCol)

val oneHotEncoder = new OneHotEncoderEstimator()
  .setInputCols(indexedCatFeatures)
  .setOutputCols(indexedCatFeatures.map(_+KeyColumns.oneHotEncoderSuffix))
  .setDropLast(false)
  // 兼容StringIndexer输出的未知特征索引，自动生成对应独热编码位
  .setHandleInvalid("keep")
val predictors = numFeatures ++ oneHotEncoder.getOutputCols
val assembler = new VectorAssembler().setInputCols(predictors).setOutputCol(KeyColumns.features)

// 把所有特征处理、模型训练逻辑整合到同一个Pipeline
val fullPipeline = new Pipeline().setStages(catIndexer ++ Array(oneHotEncoder, assembler, 你的分类器/回归器))

// 交叉验证直接使用原始训练集即可
val cv = new CrossValidator()
  .setEstimator(fullPipeline)
  .setEvaluator(new BinaryClassificationEvaluator().setLabelCol(KeyColumns.y))
  .setEstimatorParamMaps(paramGrid)
  .setNumFolds(cvConfig.folders)
  .setParallelism(cvConfig.parallelism)

val cvModel = cv.fit(trainDataset)
// 训练得到的bestModel本身就包含所有特征处理+预测逻辑，预测时直接传入原始数据即可，不会报未知特征错误
val bestModel = cvModel.bestModel.asInstanceOf[PipelineModel]
bestModel.transform(测试数据集)

2. 特殊场景：确实需要手动拼接已训练的Transformer #

如果你的业务场景必须手动组合多个已训练好的Transformer生成新Pipeline，不要用new关键字创建，根据Spark版本选择对应方法：

// Spark 3.0+ 官方推荐用法
import org.apache.spark.ml.PipelineModel
val newBestModel = PipelineModel.of(newStages)

// Spark 2.x 反射兼容写法（版本升级可能失效，非必要不推荐）
val constructor = classOf[PipelineModel].getDeclaredConstructor(classOf[String], classOf[Array[org.apache.spark.ml.Transformer]])
constructor.setAccessible(true)
val newBestModel = constructor.newInstance(bestModel.uid, newStages).asInstanceOf[PipelineModel]

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