4. Spark评估Pipeline搭建：使用Spark ML Pipeline构建标准评估流程

在金融风控场景中，模型收益评估不能依赖临时脚本或手动计算。Spark ML Pipeline 提供了一套标准化、可复用的评估流程框架，能够将数据加载、特征工程、模型推理、指标计算与异常检测串联为一条自动化流水线。本节将拆解 Pipeline 的核心组件与搭建方法，并给出生产级配置建议。

4.1 Pipeline 架构总览

一个标准的收益评估 Pipeline 包含四个阶段：数据准备、模型推理、收益计算、异常检测。每个阶段由一个或多个 Transformer 或 Estimator 组成，通过 PipelineModel 串联执行。

4.2 核心组件与代码骨架

以下代码展示了一个完整的收益评估 Pipeline 搭建过程。关键点在于使用 PipelineModel.load() 加载已训练模型，然后通过自定义 Transformer 实现收益计算与异常检测。

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.PipelineModel
import org.apache.spark.ml.Transformer
import org.apache.spark.ml.param.ParamMap
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.types.StructType

// 1. 加载已训练的模型（假设之前已保存）
val trainedModel: PipelineModel = PipelineModel.load("hdfs:///models/credit_scoring_v3")

// 2. 自定义收益计算 Transformer
class ProfitCalculator extends Transformer {
  override def transform(dataset: DataFrame): DataFrame = {
    dataset.createOrReplaceTempView("scored")
    val profitDF = dataset.sparkSession.sql("""
      SELECT
        *,
        CASE
          WHEN prediction = 1 AND label = 1 THEN -loan_amount * 0.3   -- 坏账损失
          WHEN prediction = 0 AND label = 0 THEN loan_amount * 0.05   -- 正常收益
          WHEN prediction = 1 AND label = 0 THEN -loan_amount * 0.02  -- 误杀损失
          ELSE 0
        END AS profit
      FROM scored
    """)
    profitDF
  }
  override def copy(extra: ParamMap): Transformer = defaultCopy(extra)
  override def transformSchema(schema: StructType): StructType = schema
}

// 3. 自定义异常检测 Transformer
class AnomalyDetector(threshold: Double = 0.15) extends Transformer {
  override def transform(dataset: DataFrame): DataFrame = {
    val stats = dataset.agg(
      avg("profit").alias("avg_profit"),
      stddev("profit").alias("std_profit")
    ).collect()(0)
    val avgProfit = stats.getDouble(0)
    val stdProfit = stats.getDouble(1)
    // 如果平均收益低于阈值，标记异常
    if (avgProfit < threshold) {
      println(s"[ALERT] 收益异常！平均收益: $avgProfit，标准差: $stdProfit")
    }
    dataset.withColumn("anomaly_flag", when(lit(avgProfit) < lit(threshold), lit(1)).otherwise(lit(0)))
  }
  override def copy(extra: ParamMap): Transformer = defaultCopy(extra)
  override def transformSchema(schema: StructType): StructType = schema
}

// 4. 组装 Pipeline
val evalPipeline = new Pipeline().setStages(Array(
  trainedModel,               // 阶段1：模型推理（复用已训练模型）
  new ProfitCalculator(),     // 阶段2：收益计算
  new AnomalyDetector(0.12)   // 阶段3：异常检测
))

// 5. 执行评估
val evalModel = evalPipeline.fit(trainingData)  // fit 实际只做 schema 校验
val resultDF = evalModel.transform(evalData)    // transform 执行全流程
resultDF.select("profit", "anomaly_flag").show()

4.3 组件职责与参数配置

下表列出了 Pipeline 中每个组件的核心职责、输入输出及关键参数，方便你在搭建时快速对照。

4.4 分箱收益计算：细化评估粒度

仅计算整体收益容易掩盖高风险区间的损失。建议在 Pipeline 中加入 分箱器，将预测概率分为 10 个等频区间，然后按区间计算收益指标。这样能快速定位“高损失区间”或“低收益区间”。

4.5 Pipeline 保存与版本管理

评估 Pipeline 同样需要版本化管理。每次评估后，将 PipelineModel 保存到带时间戳的路径，便于回溯和对比。

// 保存评估 Pipeline（包含模型 + 评估逻辑）
val timestamp = java.time.LocalDateTime.now().toString.replace(":", "-")
evalModel.write.overwrite().save(s"hdfs:///eval_pipelines/profit_eval_$timestamp")

// 加载历史评估 Pipeline 进行对比
val historicalEval = PipelineModel.load("hdfs:///eval_pipelines/profit_eval_2025-03-15")
val historicalResult = historicalEval.transform(evalData)

4.6 常见踩坑与规避

• Schema 不匹配：自定义 Transformer 的 transformSchema 必须返回正确的输出 Schema，否则 Pipeline 在 fit 阶段会报错。建议在开发时先打印输入 Schema 再编写输出逻辑。
• 模型与评估数据分布偏移：如果评估数据与训练数据分布差异过大，模型推理结果可能失效。建议在 Pipeline 中加入 PSI（群体稳定性指标）计算作为前置检查。
• 收益计算逻辑硬编码：损失率、收益率等参数不应硬编码在 Transformer 内部。应通过 Param 暴露为可配置参数，或从外部配置文件读取。
• 忽略缓存策略：如果 Pipeline 中多次使用同一 DataFrame（如分箱后既计算收益又计算样本量），建议在关键节点调用 .cache() 避免重复计算。