3、数据质量异常排查：缺失值检测、异常值检测、数据分布偏移检测、数据倾斜检测与处理方法

数据质量是模型稳定性的基石。在 Spark 生产环境中，数据质量异常通常表现为四种形态：缺失值、异常值、分布偏移和数据倾斜。本章将逐一拆解每种异常的检测原理、Spark 实现方法及处理策略。

3.1 缺失值检测与处理

缺失值检测的核心是识别 DataFrame 中每个字段的 null 或 NaN 比例。Spark 提供 DataFrameNaFunctions 接口进行高效统计。

3.1.1 缺失率统计

// Scala 示例：计算每列缺失率
val df = spark.read.parquet("/data/feature_store")
val totalCount = df.count()

val missingStats = df.columns.map { colName =>
  val missingCount = df.filter(df(colName).isNull || df(colName).isNaN).count()
  (colName, missingCount, missingCount.toDouble / totalCount)
}

missingStats.foreach { case (col, cnt, ratio) =>
  println(f"$col%s: 缺失 $cnt%d 条, 占比 ${ratio * 100}%.2f%%")
}

3.1.2 缺失值处理策略

策略 适用场景 Spark 实现
删除行 缺失率 < 1%，且样本充足 df.na.drop()
均值/中位数填充 连续型特征，分布较对称 df.na.fill(meanVal)
众数填充 离散型特征 df.na.fill(modeVal)
前向/后向填充 时序特征 df.withColumn("col", last("col", true).over(...))
模型预测填充 重要特征，缺失率 5%~20% 使用 Spark ML 回归/分类模型
实践建议：在风控模型中，缺失值本身可能携带信息。建议将“是否缺失”作为独立二值特征加入模型，再配合填充值，避免信息损失。

3.2 异常值检测

异常值检测的目标是识别与正常分布偏离过大的样本点。Spark 环境下常用以下三种方法。

3.2.1 基于统计的检测：IQR 与 Z-Score

// IQR 方法检测异常值
val quantiles = df.stat.approxQuantile("amount", Array(0.25, 0.75), 0.01)
val Q1 = quantiles(0)
val Q3 = quantiles(1)
val IQR = Q3 - Q1
val lowerBound = Q1 - 1.5 * IQR
val upperBound = Q3 + 1.5 * IQR

val outliers = df.filter($"amount" < lowerBound || $"amount" > upperBound)
println(s"IQR 异常值数量: ${outliers.count()}")

3.2.2 基于模型的方法：孤立森林

Spark ML 3.0+ 支持 IsolationForest，适合高维特征异常检测。

import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.iforest.{IForest, IForestModel}

val assembler = new VectorAssembler()
  .setInputCols(Array("feat1", "feat2", "feat3"))
  .setOutputCol("features")
val featureDF = assembler.transform(df)

val iforest = new IForest()
  .setNumTrees(100)
  .setMaxSamples(256)
  .setContamination(0.05)
  .setFeaturesCol("features")
  .setPredictionCol("anomaly")

val model: IForestModel = iforest.fit(featureDF)
val result = model.transform(featureDF)
result.filter($"anomaly" === 1).show()

3.2.3 异常值处理方法对比

方法 优点 缺点 Spark 适用性
截尾（Winsorize） 保留样本量，简单快速 改变分布形态 高
删除 彻底消除干扰 损失样本，可能引入偏差 高
对数/Box-Cox 变换 保留异常趋势信息 需逆变换解释 中
单独建模 对极端值单独处理 增加复杂度 低
注意：风控场景中，异常值往往是欺诈或高风险样本。删除前务必评估业务含义，建议先标记后分析，而非直接丢弃。

3.3 数据分布偏移检测

分布偏移指训练集与上线后实时数据的统计分布发生显著变化。Spark 流式作业中需持续监控。

3.3.1 PSI（群体稳定性指标）

PSI 是风控领域最常用的分布偏移指标，衡量两个分布在各分箱上的差异。

def calculatePSI(expected: DataFrame, actual: DataFrame, col: String, bins: Int = 10): Double = {
  val minVal = expected.agg(min(col)).collect()(0).getDouble(0)
  val maxVal = expected.agg(max(col)).collect()(0).getDouble(0)
  val step = (maxVal - minVal) / bins

  // 分箱统计期望分布
  val expectedBins = expected.withColumn("bin",
    floor(($"$col" - minVal) / step).cast("int"))
    .groupBy("bin").count()
    .withColumn("ratio", $"count" / sum("count").over())

  // 分箱统计实际分布
  val actualBins = actual.withColumn("bin",
    floor(($"$col" - minVal) / step).cast("int"))
    .groupBy("bin").count()
    .withColumn("ratio", $"count" / sum("count").over())

  // 计算 PSI
  val joined = expectedBins.alias("e").join(actualBins.alias("a"), "bin")
  joined.agg(sum(
    (($"a.ratio" - $"e.ratio") * log($"a.ratio" / $"e.ratio"))
  )).collect()(0).getDouble(0)
}

3.3.2 分布偏移阈值与响应

PSI 值 偏移程度 建议操作
< 0.1 无显著偏移 继续监控
0.1 ~ 0.25 轻度偏移 触发告警，检查特征来源
> 0.25 严重偏移 暂停模型，重新训练或回滚
关键点：分布偏移检测应纳入 Spark Structured Streaming 的 foreachBatch 中，每批次计算 PSI，实现实时监控。

3.4 数据倾斜检测与处理

数据倾斜是 Spark 作业性能下降的首要原因。倾斜表现为某些分区数据量远大于其他分区，导致长尾任务。

3.4.1 倾斜检测方法

// 检测各分区数据量
val partitionSizes = df.rdd.mapPartitionsWithIndex { (idx, iter) =>
  Iterator((idx, iter.length))
}.collect()

partitionSizes.sortBy(-_._2).take(10).foreach { case (pid, size) =>
  println(f"分区 $pid%d: $size%d 条")
}

// 计算倾斜系数（最大分区 / 平均分区）
val avgSize = partitionSizes.map(_._2).sum / partitionSizes.length
val maxSize = partitionSizes.map(_._2).max
val skewRatio = maxSize.toDouble / avgSize
println(f"倾斜系数: $skewRatio%.2f （阈值 > 3 视为严重倾斜）")

3.4.2 倾斜处理策略

策略 原理 代码示例
加盐（Salting） 对倾斜 key 添加随机前缀，分散到多个分区 df.withColumn("salt", rand() * 10).withColumn("new_key", concat($"key", lit("_"), $"salt"))
调整并行度 增大 shuffle 分区数 df.repartition(200, $"key")
广播小表 将小表广播避免 shuffle df.join(broadcast(smallDF), "key")
两阶段聚合 先局部聚合，再全局聚合 df.withColumn("salt", rand() * 10).groupBy("salt", "key").agg(...).groupBy("key").agg(...)
最佳实践：在 join 前先对两表做 skewJoinHint（Spark 3.4+），或使用 Adaptive Query Execution (AQE) 自动处理倾斜。开启 AQE：spark.sql.adaptive.enabled=true。
数据质量异常检测与处理全景流程 原始数据输入 缺失值检测 异常值检测 分布偏移检测 数据倾斜检测 删除 / 均值填充 前向填充 / 模型预测 IQR / Z-Score 孤立森林 / 截尾 PSI 计算 分箱对比 / 流式监控 加盐 / 调整并行度 广播小表 / AQE 清洗后高质量数据 持续监控与告警 反馈重训练

3.5 综合排查流程

在实际 Spark 模型风险排查中，数据质量异常往往同时出现。建议按以下顺序执行排查流水线：

  1. 第一步：运行缺失率统计，标记缺失率 > 30% 的特征为高风险。
  2. 第二步：对连续型特征执行 IQR 异常检测，输出异常样本清单。
  3. 第三步：计算每个特征的 PSI，对比训练集与最近 7 天生产数据。
  4. 第四步：检查 Spark UI 中 Stage 任务耗时分布，识别倾斜分区。
  5. 第五步：根据检测结果，依次执行填充、截尾、加盐等处理。
  6. 第六步：将处理后的数据写入新的特征表，并更新模型监控基线。
自动化建议：将上述流程封装为 Spark 作业，使用 spark.sql.adaptive.coalescePartitions.enabled=true 和动态资源分配，实现数据质量自愈。
数据倾斜检测与处理详细流程 检测倾斜分区 识别倾斜 Key 倾斜系数 > 3? 是 否 严重倾斜处理策略 • 加盐：对倾斜 Key 添加随机前缀 • 两阶段聚合：局部聚合 + 全局聚合 • 广播小表：避免 Shuffle 轻度倾斜处理 • 增加分区数 • 启用 AQE 自动倾斜 • 调整 spark.sql.shuffle.partitions 均衡分区数据，作业性能恢复 验证：检查 Spark UI 长尾任务是否消除

3.6 总结

数据质量异常排查是 Spark 模型风险防控的第一道防线。本章覆盖了四种核心异常类型的检测与处理方法：

  • 缺失值：使用 na.drop/na.fill 快速处理，注意保留缺失指示特征。
  • 异常值：IQR 适合快速筛查，孤立森林适合高维场景。
  • 分布偏移：PSI 是标准指标，需在流式作业中持续计算。
  • 数据倾斜：加盐和 AQE 是最高效的解决方案。

建议将上述检测逻辑封装为 Spark 监控作业，每日定时运行并输出报告，实现数据质量的自动化巡检。

← 上一章 📖 返回目录 下一章 →