第35课：毕业项目：端到端ML Pipeline

1. 端到端ML Pipeline

数据获取→数据验证→特征工程→模型训练→模型验证→模型部署→性能监控→反馈循环

2. 数据管理

DVC(数据版本)、Great Expectations(数据质量)、Feature Store(共享特征)

3. 模型训练

实验跟踪(MLflow/W&B)、超参数优化(Optuna/Bayesian)、模型选择策略(快基线→增复杂度→集成→微调)

4. 模型部署

序列化：Pickle/Joblib/ONNX。容器化：Docker+K8s。

5. 模型监控

性能漂移(准确率下降)、数据漂移(PSI>0.25=显著)、概念漂移(P(Y|X)变化)

重训练触发：定期/性能低/漂移显著

6. MLOps成熟度

L0(手动)→L1(Pipeline自动化)→L2(CI/CD)→L3(持续训练全自动)

7. 课程总结

继续学习: 深度学习(CNN/RNN/Transformer)、强化学习、MLOps生产化

Python代码实现与验证

完整代码

import numpy as np
import pandas as pd
import time
from sklearn.datasets import fetch_california_housing
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.preprocessing import StandardScaler
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.ensemble import GradientBoostingRegressor, RandomForestRegressor
from sklearn.linear_model import Ridge
from sklearn.metrics import mean_squared_error, r2_score, mean_absolute_error
import joblib
from scipy import stats
np.random.seed(42)

housing = fetch_california_housing()
X = pd.DataFrame(housing.data, columns=housing.feature_names)
y = housing.target
Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.2, random_state=42)

preproc = ColumnTransformer([('num', Pipeline([('imp',SimpleImputer(strategy='median')),('s',StandardScaler())]), X.columns)])

print("=== 端到端ML Pipeline ===")
print("1.数据获取: {} {}".format(X.shape, '无缺失' if X.isnull().sum().sum()==0 else '有缺失'))

results = {}
for name,mdl in [('Ridge',Ridge(1)),('RF',RandomForestRegressor(n_estimators=100,random_state=42,n_jobs=-1)),
                 ('GBDT',GradientBoostingRegressor(n_estimators=200,max_depth=4,learning_rate=0.1,random_state=42))]:
    pipe = Pipeline([('p',preproc),('m',mdl)])
    t0=time.time(); pipe.fit(Xtr,ytr); t1=time.time()
    yp=pipe.predict(Xte)
    results[name]={'R2':r2_score(yte,yp),'RMSE':np.sqrt(mean_squared_error(yte,yp)),'MAE':mean_absolute_error(yte,yp),'time':t1-t0}
    print("{}: R2={:.4f} RMSE={:.4f} MAE={:.4f} ({:.1f}s)".format(name,results[name]['R2'],results[name]['RMSE'],results[name]['MAE'],results[name]['time']))

print("\n5.超参数优化")
pipe_gb = Pipeline([('p',preproc),('m',GradientBoostingRegressor(random_state=42))])
grid = GridSearchCV(pipe_gb,{'m__n_estimators':[100,200],'m__max_depth':[3,5],'m__learning_rate':[0.05,0.1]},cv=3,scoring='r2',n_jobs=-1)
grid.fit(Xtr,ytr)
print("最佳: {} CV={:.4f} Test={:.4f}".format(grid.best_params_, grid.best_score_, grid.score(Xte,yte)))

joblib.dump(grid.best_estimator_, '/tmp/best_ml_model.pkl')
loaded = joblib.load('/tmp/best_ml_model.pkl')
print("\n6.模型保存/加载: 预测一致={}".format(np.allclose(loaded.predict(Xte),grid.predict(Xte))))

print("\n7.数据漂移检测(KS检验)")
for col_idx, col in enumerate(X.columns):
    ks_p = stats.ks_2samp(Xtr[col].values if hasattr(Xtr,'columns') else Xtr[:,col_idx], Xte[col].values if hasattr(Xte,'columns') else Xte[:,col_idx]).pvalue
    if ks_p < 0.05:
        print("  {} 可能漂移 (p={:.4f})".format(col, ks_p))

print("\n=== Pipeline完成 ===")
print("全流程: 数据获取->验证->划分->预处理->训练->优化->序列化->监控")

运行结果

深入理解：从项目到产品的关键步骤

ML项目成功的关键因素

项目时间分配建议

从竞赛到生产的差距

ML系统技术债

D. Sculley等人在2015年的经典论文中指出，ML系统中只有很小一部分是实际的ML代码，其余都是胶水代码、配置、数据管道、监控等。隐藏的技术债包括：

• 数据依赖：输入特征的数据源变更
• 配置债务：超参数、特征选择等配置散落各处
• 实验债务：大量实验代码未被清理
• 应对变化：世界在变，模型需要更新

项目复盘与进阶路线

5个项目的核心收获

从竞赛到生产的差距

继续学习路线

• 深度学习：神经网络→CNN(图像)→RNN/Transformer(NLP)→大模型微调
• 强化学习：Q-Learning→Policy Gradient→PPO
• MLOps：MLflow→Docker→K8s→监控→持续训练
• 领域深耕：NLP/CV/推荐/时序/图学习

推荐学习资源