MLOps 流水线实战：模型训练到部署的自动化之路

import mlflow
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import accuracy_score, precision_score

def train_pipeline(data_path: str) -> float:
    """端到端训练流水线"""
    mlflow.set_experiment("production-model")

    with mlflow.start_run():
        # 数据加载
        df = pd.read_parquet(data_path)
        X = df.drop("target", axis=1)
        y = df["target"]

        # 划分训练集和测试集
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )

        # 模型训练
        model = GradientBoostingClassifier(
            n_estimators=300,
            max_depth=6,
            learning_rate=0.1,
        )
        model.fit(X_train, y_train)

        # 模型评估
        y_pred = model.predict(X_test)
        acc = accuracy_score(y_test, y_pred)
        prec = precision_score(y_test, y_pred, average="weighted")

        mlflow.log_metrics({"accuracy": acc, "precision": prec})
        mlflow.sklearn.log_model(model, "model")

        return acc

if __name__ == "__main__":
    accuracy = train_pipeline("s3://data/features/train.parquet")
    print(f"Model accuracy: {accuracy:.4f}")

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: production-model
spec:
  predictor:
    sklearn:
      storageUri: s3://mlflow-artifacts/1/model
      resources:
        requests:
          cpu: "1"
          memory: 2Gi
        limits:
          cpu: "2"
          memory: 4Gi