📚 课程目录第2章 Spark MLlib模型生命周期

2. Spark MLlib 模型生命周期：模型训练、保存与加载、部署、监控

在风控场景中，Spark MLlib 模型的完整生命周期涵盖从离线训练到在线推理、再到持续监控的全链路。理解每个阶段的关键操作与风险点，是进行模型风险排查的基础。本章按时间顺序拆解四个核心阶段：模型训练、模型保存与加载、模型部署、模型监控。

2.1 模型训练：Pipeline 与 Estimator

MLlib 使用 Pipeline 将特征工程与模型训练串联为一个工作流。训练阶段的核心风险包括：数据倾斜、特征穿越、参数未收敛。

  关键流程： DataFrame → Transformer（特征处理）→ Estimator（算法训练）→ PipelineModel（拟合结果）

// 示例：逻辑回归 Pipeline 训练
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.{VectorAssembler, StandardScaler}
import org.apache.spark.ml.{Pipeline, PipelineModel}

val assembler = new VectorAssembler()
  .setInputCols(Array("feature1", "feature2", "feature3"))
  .setOutputCol("rawFeatures")

val scaler = new StandardScaler()
  .setInputCol("rawFeatures")
  .setOutputCol("features")

val lr = new LogisticRegression()
  .setMaxIter(100)
  .setRegParam(0.01)

val pipeline = new Pipeline().setStages(Array(assembler, scaler, lr))
val model = pipeline.fit(trainingData)

阶段	核心组件	常见风险	排查方向
特征处理	Transformer（StringIndexer, VectorAssembler）	类别基数过高、空值未处理	检查 DataFrame 统计信息、空值比例
模型拟合	Estimator（LR, GBT, RandomForest）	不收敛、过拟合、梯度爆炸	查看 trainingLog、验证集 AUC/KS
交叉验证	CrossValidator / TrainValidationSplit	参数网格过大导致 OOM	监控 Executor 内存、调整并行度

2.2 模型保存与加载：序列化与版本管理

MLlib 使用 PipelineModel.save(path) 将完整模型持久化到 HDFS 或对象存储。加载时使用 PipelineModel.load(path)。此阶段的风险集中在元数据兼容性和文件完整性。

最佳实践： 保存时同时写入模型版本号、训练时间、特征列表等元数据到外部表（如 Hive），便于后续追溯。

// 模型保存
model.write.overwrite().save("hdfs://cluster/models/risk_lr_v1.0")

// 模型加载
val loadedModel = PipelineModel.load("hdfs://cluster/models/risk_lr_v1.0")

// 检查模型 UID 是否匹配
println(loadedModel.stages.last.uid)

操作	API	风险场景	排查方法
保存	write.overwrite().save()	写入中途失败导致目录不完整	检查 _SUCCESS 标记文件；对比文件数
加载	PipelineModel.load()	Spark 版本不兼容（2.x vs 3.x）	检查 metadata/part-00000 中的 sparkVersion
版本管理	手动维护版本表	模型文件被误覆盖	启用 HDFS 回收站或对象存储版本控制

2.3 模型部署：批处理与实时推理

风控模型部署通常有两种模式：离线批处理（T+1 评分）和实时 API（毫秒级决策）。MLlib 原生模型适合批处理；实时场景需转换为 PMML 或 ONNX 格式，或使用 Spark 的 transform 方法在微批中执行。

常见陷阱： 实时部署时直接使用 PipelineModel.transform() 处理单条数据，会导致 Driver 端序列化开销极大。应使用 mapPartitions 或广播模型变量。

// 批处理部署：对全量用户评分
val scoreDF = loadedModel.transform(batchData)
scoreDF.select("user_id", "prediction", "probability").show()

// 实时部署优化：广播模型
val broadcastModel = spark.sparkContext.broadcast(loadedModel)
val resultRDD = inputRDD.mapPartitions { partition =>
  val model = broadcastModel.value
  partition.map { row => model.transform(row) }
}

2.4 模型监控：漂移检测与性能衰减

部署后的模型需要持续监控两个维度：数据漂移（特征分布变化）和概念漂移（标签分布变化）。MLlib 不提供内置监控，通常结合 Spark Streaming 或外部工具（如 Evidently、Whylogs）实现。

  监控指标清单：
  PSI（群体稳定性指标）：衡量特征分布偏移
KS / AUC：区分能力是否衰减
空值率 / 异常值率：数据质量变化
模型评分均值：整体风险偏好是否偏移

// 使用 Spark 计算 PSI 示例
val expected = spark.table("expected_features")  // 训练集分布
val actual = spark.table("actual_features")      // 当前批次分布

val psiDF = expected.join(actual, "feature_bin")
  .withColumn("psi_component", 
    ($"actual_pct" - $"expected_pct") * log($"actual_pct" / $"expected_pct"))
  .groupBy("feature_name")
  .agg(sum("psi_component").alias("psi"))

psiDF.filter($"psi" > 0.25).show()  // 超过阈值触发告警

监控维度	常用指标	告警阈值	排查动作
数据漂移	PSI、KL 散度	PSI > 0.25	检查上游特征源、ETL 逻辑变更
概念漂移	KS、AUC 衰减率	AUC 下降 > 5%	重新训练或回滚至稳定版本
数据质量	空值率、异常值率	空值率 > 10%	排查数据管道、补全缺失字段

风险排查建议： 在模型监控阶段，优先对比当前批次与训练批次的 summary 统计量（均值、方差、分位数）。若发现某个特征均值偏移超过 3 个标准差，应立即触发告警并阻断模型自动决策。