3、实时数据清洗：数据质量检查、缺失值处理、异常值检测与过滤、数据去重策略

实时数据清洗，说白了就是给流进来的数据「洗澡」。数据从源头过来，什么脏东西都有——字段丢了、值跑偏了、重复记录扎堆了。如果不处理，模型训练出来就是个「垃圾进，垃圾出」的结局。

我刚开始做实时特征工程时，总觉得清洗是小事。结果有一次线上模型突然掉点，查了两天才发现是某个传感器持续上报了 NaN 值，模型直接学歪了。从那以后，我把数据清洗提到了流水线最前面。

3.1 数据质量检查

数据质量检查是清洗的第一道关卡。我习惯把它分成两个层面：结构检查和内容检查。

3.1.1 结构检查

说白了就是看数据长没长「歪」。字段名对不对、类型匹不匹配、有没有多出或少掉字段。

举个例子，你期望收到的是 user_id: string, age: int, score: float，结果上游突然把 age 发成了字符串，或者干脆漏了 score 字段。这种问题在实时流里特别常见，尤其是多个数据源合并的时候。

// 伪代码：实时 Schema 校验
class SchemaValidator:
    expected_schema = {"user_id": str, "age": int, "score": float}
    
    def validate(record):
        for field, expected_type in expected_schema.items():
            if field not in record:
                return False, f"缺少字段: {field}"
            if not isinstance(record[field], expected_type):
                return False, f"类型不匹配: {field}"
        return True, "OK"

3.1.2 内容检查

结构对了，内容也可能有问题。比如 age 字段是 int 类型，但值是 -5 或者 200。这种逻辑上的不合理，结构检查是抓不到的。

我一般会维护一个值域规则表，在实时流里做范围校验。

字段	规则	违规处理
age	0 < age < 150	丢弃并告警
score	0.0 <= score <= 100.0	截断到边界值
email	正则匹配邮箱格式	标记为无效

3.2 缺失值处理

缺失值在实时流里几乎是躲不掉的。传感器掉线、网络抖动、上游系统故障，都会导致字段为空。

处理缺失值，我一般有三种策略，按优先级排序：

1. 实时填充：用滑动窗口内的均值、中位数或最近值填充。适合数值型字段。
2. 标记缺失：保留原始值，但增加一个 is_missing 标志位。适合模型能处理缺失的场景。
3. 丢弃记录：如果关键字段缺失，直接丢弃。适合对数据完整性要求极高的场景。

// 滑动窗口均值填充
class SlidingMeanImputer:
    def __init__(self, window_size=1000):
        self.window = deque(maxlen=window_size)
    
    def impute(self, value):
        if value is None or math.isnan(value):
            if len(self.window) == 0:
                return 0.0  # 默认值
            return sum(self.window) / len(self.window)
        else:
            self.window.append(value)
            return value

我曾经遇到过一个场景：某个 IoT 项目的温度传感器每 5 秒上报一次数据，但网络偶尔抖动导致 10% 的数据包丢失。如果用全局均值填充，温度曲线会变得很平，完全失去实时性。后来改用最近值填充，效果好了很多。

3.3 异常值检测与过滤

异常值检测，说白了就是找出那些「不对劲」的数据点。在实时场景下，我常用的方法有三种：

3.3.1 统计方法

最经典的是 3σ 原则和 IQR（四分位距）。3σ 适合正态分布的数据，IQR 更鲁棒，不受极端值影响。

// IQR 异常检测
def is_outlier_iqr(value, recent_values):
    q1 = np.percentile(recent_values, 25)
    q3 = np.percentile(recent_values, 75)
    iqr = q3 - q1
    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr
    return value < lower_bound or value > upper_bound

3.3.2 模型方法

如果数据分布复杂，我会用轻量级的在线模型，比如 Isolation Forest 的流式版本，或者 Autoencoder 的重构误差。不过要注意，模型推理必须在毫秒级完成。

3.3.3 规则方法

有些异常是业务规则就能判定的。比如用户下单金额不能超过 100 万，或者传感器数值变化率不能超过 50%。这种规则简单高效，适合放在 pipeline 最前面。

3.4 数据去重策略

数据去重在实时流里是个老大难问题。为什么？因为流数据没有「全局唯一性」的概念。同一条数据可能因为网络重传、消费者重启、上游重试等原因，出现多次。

我常用的去重策略有四种：

策略	原理	适用场景	缺点
基于 ID 的去重	维护一个已处理 ID 集合	数据有唯一 ID	集合无限增长
滑动窗口去重	只保留窗口内的 ID	允许短时间重复	窗口外可能重复
布隆过滤器	概率性去重	高吞吐、容忍误判	有误判率
状态后端去重	用 RocksDB 等持久化状态	需要精确去重	性能开销大

我个人最常用的是布隆过滤器 + 滑动窗口的组合。布隆过滤器负责快速判断「这个 ID 见过没」，滑动窗口负责定期清理过期 ID，防止内存爆炸。

// 布隆过滤器 + 滑动窗口去重
class Deduplicator:
    def __init__(self, window_sec=300, expected_inserts=100000):
        self.bloom = BloomFilter(expected_inserts, 0.01)  # 1% 误判率
        self.window = {}  # {id: timestamp}
        self.window_sec = window_sec
    
    def is_duplicate(self, id, timestamp):
        # 清理过期 ID
        self._evict_expired(timestamp)
        
        if id in self.bloom:
            return True
        self.bloom.add(id)
        self.window[id] = timestamp
        return False
    
    def _evict_expired(self, current_ts):
        expired = [id for id, ts in self.window.items() 
                   if current_ts - ts > self.window_sec]
        for id in expired:
            del self.window[id]
            # 注意：布隆过滤器不支持删除，只能等重建

3.5 清洗流水线的整体编排

好了，四个核心步骤都讲完了。那它们怎么串起来呢？我一般这样编排：

1. 第一步：Schema 校验 — 结构不对的直接丢弃
2. 第二步：去重 — 避免后续步骤处理重复数据
3. 第三步：缺失值处理 — 填充或标记
4. 第四步：异常值检测 — 过滤或修正

这个顺序是有讲究的。去重要放在前面，因为重复数据会干扰统计窗口的计算。缺失值处理要放在异常值检测前面，因为缺失值本身不是异常，但如果不处理，后续的统计方法会把 NaN 当成异常。