机器学习入门：热更新 数据预处理

最近在折腾一个小模型服务，模型热更新倒是能用了，结果数据预处理这块卡住了。线上字段加了一个类别，预处理代码里的映射是训练时 fit 出来的，服务重启一次还能忍，但每次改映射都重启就太蠢了。于是想着能不能热更新预处理。一开始以为 watch 一下 yaml 重新加载就完事，结果发现这玩意儿坑不少，可能是我没找对姿势吧。

先说结论：预处理热更新不是 reload 一下代码

数据预处理跟模型不一样。模型权重是参数，预处理里也有一半是“参数”：均值方差、缺失值填充值、类别映射、特征顺序、分箱边界。这些必须在训练集上 fit，推理时只能 transform。热更新时如果直接把新配置读进来，很容易出现特征顺序错、类别编码对不上、标准化用了新均值但模型还是旧模型，预测直接飘。所以第一件事不是写监听，是把预处理拆成“配置 + 固定变换代码”。配置里只放 transform 需要的参数，不放训练逻辑。

把预处理配置拆出来

我现在的做法是一个 yaml 管推理预处理，类似这样：

version: 20240521-1
model_version: yolo-nas-s-v3
features:
  - name: age
    type: numeric
    fill: 30
    mean: 35.2
    std: 8.1
  - name: city
    type: category
    fill: "__UNK__"
    mapping:
      beijing: 0
      shanghai: 1
      "__UNK__": 2

推理时只按这个文件做 transform，fit 的结果在训练脚本里导出，不在这里现算。这样热更新就是换配置，不碰训练数据。需要加字段就加字段，但注意模型输入维度别乱变，除非模型也一起换。

监听文件并原子切换

监听用 watchdog 就行，别自己写轮询。关键不是“文件变了就 reload”，而是“读完、校验、再切换”。我一开始直接在回调里 open 文件，结果编辑器保存时先清空再写入，读到半个 json，服务直接报错。后来改成先读临时变量，校验通过再赋值，并且用读写锁包一下。最小代码大概这样：

import json
import threading
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class PreprocessState:
    def __init__(self, path):
        self.path = path
        self.lock = threading.RLock()
        self.config = self._load()
        self.version = self.config["version"]

    def _load(self):
        with open(self.path, "r", encoding="utf-8") as f:
            cfg = json.load(f)
        assert "version" in cfg and "features" in cfg
        return cfg

    def reload(self):
        new_cfg = self._load()
        with self.lock:
            self.config = new_cfg
            self.version = new_cfg["version"]

    def transform(self, row):
        with self.lock:
            cfg = self.config
        out = []
        for feat in cfg["features"]:
            v = row.get(feat["name"], feat.get("fill"))
            if feat["type"] == "numeric":
                v = (float(v) - feat["mean"]) / feat["std"]
            else:
                v = feat["mapping"].get(v, feat["mapping"]["__UNK__"])
            out.append(v)
        return out


class Handler(FileSystemEventHandler):
    def __init__(self, state):
        self.state = state

    def on_modified(self, event):
        if event.src_path.endswith("preprocess.json"):
            time.sleep(0.2)
            self.state.reload()

这里 time.sleep(0.2) 很土，但能躲过一部分半写入。更稳的是写临时文件再 os.replace 原子替换，生产环境建议直接上配置中心，别跟文件系统较劲。

预处理代码本身怎么热更新

如果连 transform 逻辑都要热更新，那就不是改配置了，是改代码。Python 里 reload 模块听着简单，实际坑很多：旧对象还引用老函数、类定义不一致、线程里跑着旧逻辑。入门阶段我不建议一上来搞代码热更新，先把参数热更新做稳。真要改逻辑，用策略注册表，把每个预处理版本注册成函数，配置里写 pipeline: v2，切换时只换函数引用。新函数先离线跑一批样本对比，别直接上。

踩过的坑

第一个坑是版本绑定。模型文件和预处理配置得一起换，不然新模型用旧均值，loss 看着正常，线上指标就是不对。我现在在配置里写 model_version，加载时校验，不匹配就拒绝切换。第二个坑是缓存。有些服务会缓存 transform 结果，热更新后缓存没清，等于没更新。第三个坑是并发，一边读配置一边替换，轻则读到旧值，重则读到半截。锁不贵，别省。

怎么验证

热更新完别只看日志说“reload success”。拿几条固定样本，旧配置和新配置各跑一遍，看输出差异是不是符合预期；再拿模型做一次预测，确认输入维度、特征顺序没变。最后留一个回滚按钮，配置文件保留上一版，出问题切回去。机器学习入门阶段，热更新不是炫技，是让预处理别成为每次改字段都要重启的理由。先跑起来，再慢慢补监控和回滚，不然一开始就想搞大而全，基本会卡在配置上。