AI数据工程不是"洗洗数据"：从采集到特征仓库的工程化流水线，附可运行代码

# contracts.py
from pydantic import BaseModel, Field, field_validator
from datetime import datetime

class RawEvent(BaseModel):
    user_id: str = Field(min_length=1)
    item_id: str = Field(min_length=1)
    event_type: str
    ts: datetime
    features: dict = Field(default_factory=dict)

    @field_validator("event_type")
    @classmethod
    def valid_type(cls, v):
        assert v in {"click","view","buy","fav"}, f"非法事件类型:{v}"
        return v

    @field_validator("ts")
    @classmethod
    def not_future(cls, v):
        assert v <= datetime.utcnow(), "时间戳不能是未来"
        return v

# ingest.py
import json, hashlib
from contracts import RawEvent

def ingest(line: str, sink):
    try:
        evt = RawEvent(**json.loads(line))
    except Exception as e:
        sink.dlq(line, str(e)); return
    # 幂等键：用户+物品+事件+时间戳哈希，重复消费不重复写
    key = hashlib.sha1(f"{evt.user_id}{evt.item_id}{evt.event_type}{evt.ts}".encode()).hexdigest()
    sink.upsert(key, evt.model_dump())

# clean.py
import pandas as pd

def clean(df: pd.DataFrame) -> pd.DataFrame:
    df = df.drop_duplicates(subset=["user_id","item_id","event_type","ts"])
    # 规则：异常时长、缺失关键字段
    df = df[df["features"].apply(lambda x: x.get("dwell_ms",0) < 3600_000)]
    df = df.dropna(subset=["user_id","item_id"])
    # 语义去重：文本特征做minhash近似去重（伪代码示意）
    # from datasketch import MinHashLSH
    return df.reset_index(drop=True)

# features.py
from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, Int64
from datetime import timedelta

user = Entity(name="user_id", join_keys=["user_id"])
src = FileSource(path="data/features.parquet", timestamp_field="event_ts")

user_fv = FeatureView(
    name="user_behavior",
    entities=[user],
    ttl=timedelta(days=7),
    schema=[
        Field(name="click_cnt_7d", dtype=Int64),
        Field(name="buy_rate_30d", dtype=Float32),
        Field(name="avg_dwell_ms", dtype=Float32),
    ],
    source=src,
    online=True,   # 关键：同一份定义同时产出离线和在线特征
)

# versioning.py
import dvc.api

def train_dataset(path="data/train.parquet"):
    # DVC跟踪数据版本，git commit关联数据hash
    with dvc.api.open(path, repo=".", rev="v1.2.0") as f:
        return load(f)

def log_lineage(dataset_hash, feature_view_ver, model_ver):
    lineage = {"dataset": dataset_hash, "feature_view": feature_view_ver, "model": model_ver}
    # 写入元数据库，任何一次模型故障都能反查是哪版数据、哪版特征
    db.lineage.insert_one(lineage)

# monitor.py
from scipy.stats import ks_2samp

def drift_check(ref, cur, cols, thr=0.05):
    alerts = []
    for c in cols:
        stat, p = ks_2samp(ref[c].dropna(), cur[c].dropna())
        if p < thr:
            alerts.append({"col": c, "ks": round(stat,3), "p": round(p,4)})
    return alerts  # 空列表表示无显著漂移