2、Spark评分模型基础：Spark MLlib简介、DataFrame与RDD在评分建模中的应用、特征工程与Spark Pipeline

2.1 Spark MLlib 概述与催收评分场景适配

Spark MLlib 是 Apache Spark 生态中原生的分布式机器学习库，专为大规模数据处理与模型训练设计。在催收评分场景中，MLlib 的核心价值在于：

• 海量样本处理：支持千万级逾期客户记录的全量训练，无需采样。
• 端到端流水线：从特征加工到模型评估，统一在 Spark 集群内完成。
• 生产级稳定性：与 Spark 核心引擎深度集成，适合每日/每周评分模型重训。

2.2 DataFrame 与 RDD 在评分建模中的角色

Spark 提供两种核心数据抽象：RDD（弹性分布式数据集）和 DataFrame。在催收评分建模中，两者的定位与使用场景有明确分工。

2.2.1 RDD：底层计算引擎

RDD 是 Spark 最底层的分布式数据集合，提供细粒度的转换操作（map、filter、reduceByKey）。在评分建模中，RDD 主要用于：

• 自定义复杂特征逻辑（如按客户 ID 聚合逾期天数序列）。
• 非结构化数据解析（如催收通话录音元数据）。
• 与旧版 Spark 代码兼容。

2.2.2 DataFrame：结构化建模首选

DataFrame 在 RDD 之上增加了 Schema 信息和 Catalyst 优化器，是 MLlib 评分建模的标准数据载体。其优势包括：

• 自动优化：Catalyst 优化器自动选择最佳执行计划。
• 丰富 API：支持 SQL 风格操作（select、groupBy、join）。
• MLlib 原生支持：Pipeline 组件直接接受 DataFrame 作为输入。

// 示例：RDD 转 DataFrame 用于 MLlib 训练
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

val rdd = sc.parallelize(Seq(
  ("C001", 30, 1.0),
  ("C002", 45, 0.0)
))

val schema = StructType(Seq(
  StructField("customer_id", StringType, true),
  StructField("overdue_days", IntegerType, true),
  StructField("label", DoubleType, true)
))

val df = spark.createDataFrame(rdd.map(r => Row(r._1, r._2, r._3)), schema)
df.show()

2.3 特征工程与 Spark Pipeline

催收评分模型的特征工程通常包含：缺失值填充、类别编码、数值分箱、特征缩放、特征选择等步骤。Spark MLlib 通过 Pipeline 机制将这些步骤串联为可复用的工作流。

2.3.1 核心 Pipeline 组件

2.3.2 催收特征工程典型步骤

1. 缺失值处理：使用 Imputer 用中位数或众数填充逾期金额、联系次数等字段。
2. 类别特征编码：StringIndexer + OneHotEncoder 处理催收地区、产品类型。
3. 连续特征分箱：QuantileDiscretizer 将逾期天数分为 0-30、31-60、61-90 等区间。
4. 特征缩放：MinMaxScaler 或 StandardScaler 统一量纲，加速模型收敛。
5. 特征选择：ChiSqSelector 基于卡方检验筛选 Top-K 强相关特征。

2.3.3 完整 Pipeline 代码示例

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature._
import org.apache.spark.ml.classification.LogisticRegression

// 1. 定义特征工程阶段
val indexer = new StringIndexer()
  .setInputCol("product_type")
  .setOutputCol("product_idx")
  .setHandleInvalid("keep")

val encoder = new OneHotEncoder()
  .setInputCol("product_idx")
  .setOutputCol("product_vec")

val imputer = new Imputer()
  .setInputCols(Array("overdue_days", "contact_count"))
  .setOutputCols(Array("overdue_days_imp", "contact_count_imp"))
  .setStrategy("median")

val scaler = new MinMaxScaler()
  .setInputCol("overdue_days_imp")
  .setOutputCol("overdue_days_scaled")

// 2. 定义模型
val lr = new LogisticRegression()
  .setLabelCol("label")
  .setFeaturesCol("features")
  .setMaxIter(50)

// 3. 组装 Pipeline
val pipeline = new Pipeline()
  .setStages(Array(indexer, encoder, imputer, scaler, lr))

// 4. 训练
val pipelineModel = pipeline.fit(trainingData)

// 5. 预测
val predictions = pipelineModel.transform(testData)

2.4 催收评分 Pipeline 架构图

2.5 催收评分特征工程关键对比