Python · 序列化: pickle / JSON / msgpack

场景 1 · ml 模型 + 预处理管道: joblib 整体落盘

模型上线要保证"标准化器 + 特征工程 + 分类器"作为一个整体原子保存, 单独存权重会版本错配:

import joblib
from sklearn.pipeline import Pipeline

pipe = Pipeline([
    ("scaler", StandardScaler()),            # 预处理与模型同进同出
    ("clf",    GradientBoostingClassifier()),
]).fit(X_train, y_train)

joblib.dump(pipe, "models/churn_v7.joblib", compress="lz4")
# 服务端加载: 与训练时同版本 sklearn, 否则 unpickle 可能炸
pipe = joblib.load("models/churn_v7.joblib")
prob = pipe.predict_proba(feats)[0, 1]

模型文件名带版本 + 元数据登记 sklearn 版本: 线上模型与训练环境可追溯、可回滚。

场景 2 · Redis 缓存层 pickle 存对象, 版本号 key 防结构漂移

用户画像对象每请求重建要 300ms, 想整对象缓存; 但直接 pickle 裸 key 上线, 改一次类定义缓存就静默错乱:

import pickle, redis

r = redis.Redis()
SCHEMA = "profile:v3"                # 类结构一变就 bump: v3→v4

def get_profile(uid: int) -> Profile:
    key = f"{SCHEMA}:{uid}"
    if raw := r.get(key):                  # Redis 出入都是 bytes, 正配 pickle
        return pickle.loads(raw)          # 来源=自己写入, 可信通道
    p = build_profile(uid)                 # 300ms 的重建
    r.setex(key, 3600, pickle.dumps(p, protocol=5))
    return p
# 事故教训: 不带版本号时, 旧 schema 的缓存命中后字段缺失, 线上 nil 错乱无感知

场景 3 · API 响应 orjson 替换 json: 列表接口提速

商品列表 4MB JSON 响应, 序列化占 80ms; 换 orjson 的 ORJSONResponse 三行收工:

from fastapi.responses import ORJSONResponse
from fastapi import FastAPI

app = FastAPI(default_response_class=ORJSONResponse)

@app.get("/skus", response_class=ORJSONResponse)
def list_skus():
    return {"items": sku_repo.all()}   # datetime/UUID 直接吃, 不用 default

# 压测: 4MB 响应序列化 82ms → 11ms; P99 340ms → 265ms
# 注意: ORJSONResponse 输出紧凑无缩进, 调试期想看格式再手动 orjson.dumps(..., option=OPT_INDENT_2)

场景 4 · 自定义 JSON 编码器: datetime/Decimal/UUID 统一口径

接口里日期格式有的带 Z 有的不带、金额直接 float 漂移一分钱; 出口统一走一个 default:

import json
from datetime import datetime, timezone
from decimal import Decimal
from uuid import UUID

def json_default(o):
    if isinstance(o, datetime):
        return o.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    if isinstance(o, Decimal):
        return str(o)                      # 金额走字符串, 精度零漂移
    if isinstance(o, UUID):
        return str(o)
    raise TypeError(f"not serializable: {type(o)}")  # 别静默吞

out = json.dumps(order, default=json_default, ensure_ascii=False)

场景 5 · 对外接口 pydantic 白名单: 内部字段不得出门

User ORM 对象直接 json 化返回, 把 password_hash/is_internal 字段一起漏了出去; 出口模型只放白名单:

from pydantic import BaseModel, ConfigDict

class UserPublic(BaseModel):
    model_config = ConfigDict(extra="forbid")    # 多余字段直接报错
    uid: int
    name: str
    created_at: datetime

def user_view(u: UserORM) -> str:
    # 从 ORM 挑选字段构造 → password_hash 之类根本进不来
    return UserPublic.model_validate(u, from_attributes=True).model_dump_json()

# 内部对象随便长, 出口只认视图模型: "序列化面"即攻击面

场景 6 · MQ 消息体 msgpack: 二进制降带宽

行情推送 5 万条/秒, JSON 文本 1.8KB/条把带宽打满; 换 msgpack:

import msgpack

def publish(quote: dict):
    body = msgpack.packb(quote, use_bin_type=True)   # str/bytes 严格区分
    mq.send(routing_key="quote", body=body, content_type="application/msgpack")

def consume(body: bytes):
    return msgpack.unpackb(body, raw=False)      # raw=False: str 按 str 还原

# 实测: 平均 1.8KB → 0.9KB, 网卡出流量减半;
# 嵌套数字数组 (行情快照) 收益最大, 比 JSON 小 50%+

场景 7 · 配置快照用 JSON 不用 pickle

风控参数要留档审计、diff 出变更单; pickle 是黑盒字节, JSON 一眼可读:

def snapshot(cfg: RiskConfig, path):
    with open(path, "w", encoding="utf-8") as f:
        json.dump(cfg.model_dump(), f, ensure_ascii=False,
                  indent=2, sort_keys=True)   # 排序=稳定 diff

# git diff 两个快照: 参数谁改了、何时改的一目了然
# pickle 快照: 无法 diff、无法 review、还绑着类路径 —— 审计场景禁用

场景 8 · 跨版本数据文件演进: __setstate__ 迁移旧字段

训练产物 FeatureStore v1 落盘的 pickle, 类升级到 v3 后 loads 全炸; 在恢复钩子里做迁移:

class FeatureStore:
    VERSION = 3
    def __getstate__(self):
        return {"v": self.VERSION, "feats": self.feats, "norm": self.norm}
    def __setstate__(self, state):
        v = state.get("v", 1)                    # 旧档没有版本字段 = v1
        if v < 2: state["norm"] = default_norm()   # v1→v2: 补归一化参数
        if v < 3: state["feats"] = migrate_keys(state["feats"])
        self.__dict__.update(state)           # 迁移完再落到实例上

旧档新代码照样读; 配合"读时迁移 + 写时升版", 数据文件可以横跨数个版本存活。

场景 9 · 可信通道补充: pickle + HMAC 签名防篡改

任务队列要传复杂对象(不跨语言), pickle 最省事但怕中间被改; HMAC 给它上封条(注意: 这是防篡改, 不是加密):

import pickle, hmac, hashlib
from app.config import settings            # KEY 走 K8s Secret 注入

KEY = settings.JOB_HMAC_KEY                  # 双端共享, 绝不入库入仓

def seal(obj) -> bytes:
    body = pickle.dumps(obj, protocol=5)
    sig = hmac.new(KEY, body, hashlib.sha256).digest()
    return sig + body                       # 前 32 字节 = 签名

def unseal(blob: bytes):
    sig, body = blob[:32], blob[32:]
    expect = hmac.new(KEY, body, hashlib.sha256).digest()
    if not hmac.compare_digest(sig, expect):   # 恒时比较, 防时序侧信道
        raise TamperedJobError("signature mismatch")
    return pickle.loads(body)             # 验签通过才 unpickle

场景 10 · 大 DataFrame 选型: parquet 落盘, feather 中转

特征工程 200 万行 × 380 列, 有人用 pickle 整存: 文件 1.9GB、读回 40s、只能 Python 读; 换列存:

df.to_parquet("features/v7.parquet", compression="zstd")
# 1.9GB → 420MB (zstd 列压缩), 读回 3.2s, 按需只读部分列 (columns=[...])
# 跨语言: Spark/Arrow/Rust 都能读 —— 数据不再绑死 Python

df.to_feather("tmp/scratch.feather")
# 流水线中间缓存: mmap 零拷贝, 下游进程秒级拿到同一份内存
# 一句话: 归档/共享 parquet, 短命中中间态 feather, pickle 不进数据管道