3. 数据漂移检测基础：统计检验方法在Spark中的实现原理

数据漂移检测是ML模型监控的第一道防线。当生产数据的分布与训练数据分布发生显著偏离时，模型性能往往会随之下降。本章聚焦三种最常用的统计检验方法——KS检验、卡方检验和JS散度，并深入讲解它们在Spark分布式环境下的实现原理与工程化要点。

3.1 三种统计检验方法的核心差异

在开始Spark实现之前，需要明确每种方法适用的数据类型和检测目标。下表从多个维度进行了对比：

3.2 KS检验在Spark中的实现原理

KS（Kolmogorov-Smirnov）检验通过比较两个样本的累积分布函数（CDF）之间的最大垂直距离来判断它们是否来自同一分布。在Spark中，实现KS检验的核心挑战在于：如何在分布式数据集上高效计算经验CDF。

3.2.1 分布式经验CDF计算

Spark的分布式计算模型决定了我们不能像单机那样直接对所有数据排序后计算CDF。标准做法是使用分位数近似策略：

1. 全局排序与分桶：通过sortWithinPartitions在每个分区内排序，然后使用approxQuantile获取全局分位数边界。
2. 直方图近似：将连续值域划分为K个等宽或等频桶，统计每个桶内的样本计数，用桶的累积频率近似CDF。
3. 并行计算D统计量：在分桶后的数据上，逐桶计算训练集与生产集累积频率的绝对差，取最大值作为KS统计量。

// Spark Scala实现KS检验的核心逻辑（伪代码）
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

def computeKSTest(trainingDF: DataFrame, productionDF: DataFrame, 
                  featureCol: String, numBuckets: Int = 1000): Double = {
  
  // 1. 获取训练集和生产集的全局分位数边界（合并计算）
  val combinedDF = trainingDF.select(col(featureCol))
    .union(productionDF.select(col(featureCol)))
  
  val quantiles = combinedDF.stat.approxQuantile(featureCol, 
    Array.tabulate(numBuckets)(i => i.toDouble / numBuckets), 0.01)
  
  // 2. 定义分桶UDF
  val bucketUDF = udf((value: Double) => {
    val bucketIndex = quantiles.lastIndexWhere(q => value >= q)
    math.max(0, bucketIndex)
  })
  
  // 3. 分别对训练集和生产集进行分桶计数
  val trainBuckets = trainingDF.withColumn("bucket", bucketUDF(col(featureCol)))
    .groupBy("bucket").count()
    .withColumn("cumFreq", sum("count").over(Window.orderBy("bucket")))
    .withColumn("cdf", col("cumFreq") / sum("count").over())
  
  val prodBuckets = productionDF.withColumn("bucket", bucketUDF(col(featureCol)))
    .groupBy("bucket").count()
    .withColumn("cumFreq", sum("count").over(Window.orderBy("bucket")))
    .withColumn("cdf", col("cumFreq") / sum("count").over())
  
  // 4. 合并计算最大绝对差
  val ksStat = trainBuckets.alias("t")
    .join(prodBuckets.alias("p"), Seq("bucket"), "outer")
    .agg(max(abs(coalesce(col("t.cdf"), lit(0.0)) - 
                 coalesce(col("p.cdf"), lit(0.0)))))
    .collect()(0).getDouble(0)
  
  ksStat
}

3.3 卡方检验在Spark中的实现原理

卡方检验（Chi-Square Test）用于检验两个分类变量之间的独立性，或一个分类变量的观测频数与期望频数是否一致。在数据漂移场景中，我们通常使用卡方拟合优度检验来判断生产数据的类别分布是否与训练数据一致。

3.3.1 Spark MLlib原生支持

Spark MLlib提供了ChiSquareTest类，可以直接用于特征与标签之间的独立性检验。但对于数据漂移检测，我们需要的是两个数据集之间同一特征的分布一致性检验，因此需要手动构建列联表：

// 使用Spark SQL实现卡方检验
import org.apache.spark.sql.Row
import org.apache.spark.mllib.stat.Statistics

def chiSquareDriftTest(trainingDF: DataFrame, productionDF: DataFrame, 
                       featureCol: String): Double = {
  
  // 1. 获取训练集和生产集的类别频数
  val trainCounts = trainingDF.groupBy(featureCol).count()
    .collect().map { case Row(cat: String, cnt: Long) => (cat, cnt.toDouble) }
    .toMap
  
  val prodCounts = productionDF.groupBy(featureCol).count()
    .collect().map { case Row(cat: String, cnt: Long) => (cat, cnt.toDouble) }
    .toMap
  
  // 2. 获取所有类别并构建列联表
  val allCategories = (trainCounts.keySet ++ prodCounts.keySet).toArray.sorted
  val nTrain = trainCounts.values.sum
  val nProd = prodCounts.values.sum
  
  // 3. 计算期望频数（假设分布一致）
  val expected = allCategories.map { cat =>
    val totalCount = trainCounts.getOrElse(cat, 0.0) + prodCounts.getOrElse(cat, 0.0)
    totalCount * nTrain / (nTrain + nProd)
  }
  
  val observed = allCategories.map(cat => trainCounts.getOrElse(cat, 0.0))
  
  // 4. 计算卡方统计量
  val chi2 = observed.zip(expected).map { case (o, e) =>
    if (e > 0) math.pow(o - e, 2) / e else 0.0
  }.sum
  
  chi2
}

3.3.2 列联表构建的分布式优化

当类别数量较多（如超过1000个）时，上述实现中的collect()操作会成为性能瓶颈。优化策略包括：

• 使用crosstab方法：Spark DataFrame的stat.crosstab可以在分布式环境下直接生成列联表，避免数据拉取到Driver端。
• 布隆过滤器预筛选：对于低频类别，可以先使用布隆过滤器过滤掉训练集和生产集中出现次数均少于5次的类别，减少计算量。
• Yates连续性校正：当某个类别的期望频数小于5时，应用Yates校正避免卡方统计量膨胀。

3.4 JS散度在Spark中的实现原理

JS散度（Jensen-Shannon Divergence）是KL散度的对称化版本，具有有界性和对称性两个重要性质。对于两个概率分布P和Q，JS散度的计算公式为：

JS(P || Q) = ½ KL(P || M) + ½ KL(Q || M)，其中 M = ½ (P + Q)

3.4.1 分布式概率密度估计

在Spark中计算JS散度的核心步骤是概率密度估计。对于连续型特征，需要先进行离散化处理：

1. 自适应分桶：使用approxQuantile获取训练数据的分位数，以此作为分桶边界。这样可以保证每个桶内训练数据的概率质量大致相等，避免稀疏桶问题。
2. 概率质量计算：分别统计训练集和生产集在每个桶中的样本比例，得到两个概率向量P和Q。
3. 平滑处理：对于生产数据中出现的未见类别或空桶，使用拉普拉斯平滑（Laplace Smoothing）避免概率为零导致的数值问题。

// Spark实现JS散度（含拉普拉斯平滑）
def jsDivergence(trainingDF: DataFrame, productionDF: DataFrame,
                 featureCol: String, numBuckets: Int = 100,
                 alpha: Double = 1e-6): Double = {
  
  // 1. 基于训练数据计算分桶边界
  val quantiles = trainingDF.stat.approxQuantile(featureCol,
    Array.tabulate(numBuckets)(i => i.toDouble / numBuckets), 0.01)
  
  // 2. 分桶并计算概率质量（含平滑）
  def computeProbDist(df: DataFrame): Array[Double] = {
    val bucketExpr = when(col(featureCol) < lit(quantiles(0)), 0)
    val bucketCol = quantiles.zipWithIndex.tail.foldLeft(bucketExpr) {
      case (expr, (q, idx)) =>
        expr.when(col(featureCol) < lit(q), idx)
    }.otherwise(numBuckets - 1)
    
    val counts = df.withColumn("bucket", bucketCol)
      .groupBy("bucket").count()
      .collect().map(r => (r.getInt(0), r.getLong(1).toDouble))
      .toMap
    
    val total = counts.values.sum + alpha * numBuckets
    Array.tabulate(numBuckets)(i => 
      (counts.getOrElse(i, 0.0) + alpha) / total)
  }
  
  val p = computeProbDist(trainingDF)
  val q = computeProbDist(productionDF)
  
  // 3. 计算JS散度
  val m = p.zip(q).map { case (pi, qi) => (pi + qi) / 2.0 }
  
  val klPM = p.zip(m).map { case (pi, mi) =>
    if (pi > 0 && mi > 0) pi * math.log(pi / mi) else 0.0
  }.sum
  
  val klQM = q.zip(m).map { case (qi, mi) =>
    if (qi > 0 && mi > 0) qi * math.log(qi / mi) else 0.0
  }.sum
  
  0.5 * (klPM + klQM)
}

3.5 三种方法的Spark实现架构对比

下面的SVG图展示了三种统计检验方法在Spark中的实现架构与数据流：

3.6 阈值设定与工程实践

统计检验方法输出的统计量本身并不直接指示漂移是否发生，需要结合阈值进行判定。阈值设定有以下常见策略：

3.7 性能对比与选型决策

在Spark集群环境下，三种方法的计算性能存在显著差异。以下是在1000万条训练数据、100万条生产数据、100个特征的典型场景下的实测对比：

3.8 本章小结

• KS检验通过比较经验CDF的最大垂直距离检测分布漂移，适合连续数值型特征，Spark实现的关键是分位数近似分桶。
• 卡方检验通过比较观测频数与期望频数的差异检测类别分布漂移，Spark MLlib提供原生支持，但需注意期望频数小于5时的校正问题。
• JS散度作为KL散度的对称化版本，具有有界性和对称性优势，适合多特征综合评分，Spark实现需要重点关注概率密度估计的平滑处理。
• 三种方法在Spark中的实现都遵循分桶→统计→比较的通用模式，但分桶策略、统计量计算方式和性能特征各有不同，需要根据业务场景和特征类型合理选型。