4. 概念漂移检测基础：ADWIN、DDM、Page-Hinkley 算法原理及适用场景

在模型监控体系中，概念漂移（Concept Drift） 是导致模型预测性能随时间下降的核心原因。它指数据分布 P(X, y) 或决策边界发生改变，使得历史模型不再适用。本章聚焦三种经典且轻量的在线漂移检测算法：ADWIN、DDM 和 Page-Hinkley，剖析其数学原理、参数含义及在 Spark ML 流水线中的适用场景。

4.1 算法对比总览

4.2 ADWIN（Adaptive Windowing）算法

ADWIN 由 Albert Bifet 等人提出，核心思想是：维护一个可变长度的滑动窗口，当窗口内数据分布发生显著变化时，自动丢弃旧数据以缩小窗口，从而快速适应新分布。

算法原理

1. 维护一个窗口 W，包含最近到达的 n 个样本。
2. 将窗口划分为两个子窗口 W₀（旧）和 W₁（新），计算各自的均值 μ₀ 和 μ₁。
3. 若 |μ₀ - μ₁| > ε_cut（基于 Hoeffding 不等式计算的阈值），则判定发生漂移，丢弃 W₀ 中的旧数据。
4. 窗口自适应缩小后，继续接收新数据并重复检测。

适用场景

• 数值型特征分布监控： 如用户年龄、交易金额、传感器读数等连续特征均值发生偏移。
• 回归模型残差分析： 监控预测误差的均值是否显著偏离零。
• 分类置信度漂移： 监控模型输出概率的平均值变化，间接反映数据分布变化。

4.3 DDM（Drift Detection Method）算法

DDM 由 Gama 等人提出，专门用于分类任务中的错误率漂移检测。它假设在线学习过程中，模型预测错误率服从二项分布，通过监控错误率的均值和标准差来判断是否发生漂移。

算法原理

1. 初始化：记录第一个样本的错误率 p_min 和标准差 s_min。
2. 每到达一个新样本，更新当前错误率 p_i 和标准差 s_i。
3. 若 p_i + s_i > p_min + α · s_min，进入 告警（Warning） 状态，开始缓存数据。
4. 若 p_i + s_i > p_min + β · s_min（β > α），判定发生 漂移（Drift），重置模型并清空缓存。
5. 若在告警状态下错误率回落，则退出告警状态，继续使用当前模型。

适用场景

• 在线分类任务： 如垃圾邮件检测、欺诈交易识别、实时情感分析。
• 模型准确率监控： 当模型在最新批次数据上的准确率显著下降时触发重训练。
• 二分类与多分类均适用： 只需将预测结果与真实标签对比，计算 0/1 错误即可。

4.4 Page-Hinkley（PH）算法

Page-Hinkley 是一种序列突变检测方法，最初用于信号处理中的均值变化检测。它通过累积当前观测值与历史均值的偏差，当累积和超过阈值时判定发生漂移。

算法原理

1. 维护一个累积变量 U_n = Σ(x_i - μ₀ - δ)，其中 μ₀ 是历史均值，δ 是容忍漂移量。
2. 同时维护 m_n = min(U_1, U_2, ..., U_n)。
3. 计算 PH_n = U_n - m_n。
4. 若 PH_n > λ（阈值），判定发生漂移，重置累积和。

4.5 Spark ML 中的集成策略

在 Spark ML 流水线中，上述三种算法通常作为 漂移检测器（Drift Detector） 嵌入到推理与监控循环中。以下是一个简化的集成代码示例，展示如何在 Spark Streaming 或微批次推理中调用 ADWIN 检测器：

import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.linalg.Vectors

// 模拟 ADWIN 检测器（伪代码示意）
class ADWINDetector(delta: Double = 0.05) {
  private var window: List[Double] = List()
  private var totalMean: Double = 0.0

  def addObservation(value: Double): Boolean = {
    window = window :+ value
    // 实际 ADWIN 实现需维护子窗口统计量
    // 此处仅示意接口
    detectDrift()
  }

  private def detectDrift(): Boolean = {
    // 基于 Hoeffding 不等式判断是否漂移
    // 返回 true 表示检测到漂移
    false
  }
}

// 在 Spark Streaming 中使用
val adwin = new ADWINDetector(delta = 0.01)

streamingDataFrame.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
  val predictions = model.transform(batchDF)
  val errors = predictions.select("prediction", "label")
    .as[(Double, Double)]
    .collect()
    .map { case (pred, label) => math.abs(pred - label) }

  errors.foreach { err =>
    if (adwin.addObservation(err)) {
      println(s"Batch $batchId: 检测到概念漂移！触发模型重训练。")
      // 触发重训练逻辑
    }
  }
}

4.6 算法选择决策指南