云原生 ML

这一课你将能做到

1. 按请求模式为模型选择正确的推理形态：在线 / 批量 / 流式 / 边缘
2. 本地实现一个可运行的每日批量打分任务，给题目难度评分模型落库
3. 读懂云托管训练与 Endpoint 的通用模式（SageMaker/Vertex/AzureML），掌握 spot 训练等成本手段

为什么要学

前几课的服务形态都是「在线推理」：请求来了毫秒级返回。但 ML 的推理形态不止一种——给全站 2 万道题每天更新一次难度分，用在线服务逐题调用纯属浪费；更自然的形态是每天跑一次批量任务，把分数写进数据库供站内查询。形态选错，成本和延迟都会差一个数量级。

同时，自建 K8s 是一条重运维路线，云厂商的托管 ML 平台（SageMaker / Vertex AI / Azure ML）把训练资源、Endpoint、注册中心都打包成服务。理解它们的通用模式，你才能判断「什么时候值得上、怎么避免被锁定」。人话：形态是「怎么算」，云是「在哪算」，两件事都要按成本与延迟来决策。

先建立图像

形态	触发方式	延迟要求	本站场景	课内实践
在线 Online	用户请求实时触发	毫秒级	提交时预测是否 TLE	第 1/8/10 课的 FastAPI
批量 Batch	定时任务	分钟~小时	每日全量题目难度分	本课 batch_score.py
流式 Streaming	事件流持续驱动	秒级	实时 TLE 率特征	第 2 课 Kafka
边缘 Edge	设备本地	极低	客户端离线代码提示	TFLite（第 12 课概览）

云托管训练的通用模式（三家平台同构）：
提交代码(train.py) + 数据指针(S3/GCS) + 资源规格(instance type)
        → 平台起容器跑训练 → 指标/模型落平台存储 → 注册/部署 Endpoint
你管的：代码与数据   平台管的：环境、扩缩、容错、计费

跟着做

步骤 1：批量推理的落地任务（可运行）

题目难度评分：用历史上「有人工难度分」的题训练一个回归器，每天给新题打分。

# src/batch_score.py —— 每日批量给题目打难度分并写回数据库
import sqlite3
import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_absolute_error
from sklearn.model_selection import train_test_split
import joblib, pathlib

DB = "problems.db"

def init_db(conn: sqlite3.Connection):
    conn.executescript("""
    CREATE TABLE IF NOT EXISTS problems(
      problem_id INTEGER PRIMARY KEY, submits INTEGER,
      tle_rate_7d REAL, avg_time REAL, human_rank INTEGER, pred_rank REAL);
    """)
    conn.executemany(
        "INSERT OR IGNORE INTO problems VALUES (?,?,?,?,?,NULL)",
        [(1000 + i, 200 - i * 7, 0.5 - i * 0.02, 900 + i * 120,
          (i % 5) + 1) for i in range(20)])
    conn.commit()

def train(conn):
    labeled = pd.read_sql(
        "SELECT tle_rate_7d, avg_time, human_rank FROM problems "
        "WHERE human_rank IS NOT NULL", conn)
    X = labeled[["tle_rate_7d", "avg_time"]]
    y = labeled["human_rank"]
    X_tr, X_te, y_tr, y_te = train_test_split(X, y, random_state=42)
    model = RandomForestRegressor(n_estimators=100, random_state=42)
    model.fit(X_tr, y_tr)
    print("MAE =", round(mean_absolute_error(y_te, model.predict(X_te)), 3))
    joblib.dump(model, "models/difficulty-v1.joblib")
    return model

def score_all(conn, model):
    todo = pd.read_sql(
        "SELECT problem_id, tle_rate_7d, avg_time FROM problems", conn)
    preds = model.predict(todo[["tle_rate_7d", "avg_time"]])
    conn.executemany(
        "UPDATE problems SET pred_rank=? WHERE problem_id=?",
        zip(preds.round(2), todo["problem_id"]))
    conn.commit()

if __name__ == "__main__":
    pathlib.Path("models").mkdir(exist_ok=True)
    with sqlite3.connect(DB) as conn:
        init_db(conn)
        score_all(conn, train(conn))
        print(pd.read_sql("SELECT problem_id, human_rank, pred_rank "
                          "FROM problems LIMIT 5", conn))