MLOps：机器学习模型交付

# MLflow 实验追踪示例
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
 
# 设置 MLflow Tracking Server
mlflow.set_tracking_uri("http://mlflow.example.com:5000")
mlflow.set_experiment("orders-delay-prediction")
 
# 定义可复现的训练过程
def train_model(X_train, y_train, X_test, y_test, params):
    with mlflow.start_run(run_name=f"rf-{params['n_estimators']}") as run:
        # 记录参数
        mlflow.log_params(params)
 
        # 训练模型
        model = RandomForestClassifier(
            n_estimators=params["n_estimators"],
            max_depth=params["max_depth"],
            random_state=42
        )
        model.fit(X_train, y_train)
 
        # 评估
        y_pred = model.predict(X_test)
        metrics = {
            "accuracy": accuracy_score(y_test, y_pred),
            "f1_score": f1_score(y_test, y_pred, average="weighted"),
            "precision": precision_score(y_test, y_pred, average="weighted"),
            "recall": recall_score(y_test, y_pred, average="weighted")
        }
        mlflow.log_metrics(metrics)
 
        # 记录模型
        mlflow.sklearn.log_model(
            model,
            "model",
            registered_model_name="orders-delay-prediction"
        )
 
        # 记录数据版本
        mlflow.log_param("data_version", params.get("data_version", "unknown"))
 
        return run.info.run_id, metrics
 
# 执行训练
params = {
    "n_estimators": 100,
    "max_depth": 10,
    "data_version": "dvc://datasets/orders/v3"
}
run_id, metrics = train_model(X_train, y_train, X_test, y_test, params)
print(f"Run ID: {run_id}, Metrics: {metrics}")

# Kubeflow Pipeline 定义
from kfp import dsl
from kfp.dsl import component, Output, Model, Metrics
 
@component(base_image="python:3.11")
def data_ingestion(data_path: str, output_data: Output[dsl.Dataset]):
    import pandas as pd
    df = pd.read_csv(data_path)
    df.to_csv(output_data.path, index=False)
 
@component(base_image="python:3.11")
def data_validation(input_data: dsl.Dataset, validation_result: Output[Metrics]):
    import pandas as pd
    df = pd.read_csv(input_data.path)
    # 数据质量检查
    null_ratio = df.isnull().sum().sum() / (df.shape[0] * df.shape[1])
    validation_result.log_metric("null_ratio", null_ratio)
    validation_result.log_metric("row_count", len(df))
    validation_result.log_metric("column_count", len(df.columns))
 
@component(base_image="python:3.11")
def train_model(
    input_data: dsl.Dataset,
    n_estimators: int,
    output_model: Output[Model],
    output_metrics: Output[Metrics]
):
    import pandas as pd
    import pickle
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.model_selection import train_test_split
    from sklearn.metrics import accuracy_score, f1_score
 
    df = pd.read_csv(input_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)
 
    model = RandomForestClassifier(n_estimators=n_estimators, random_state=42)
    model.fit(X_train, y_train)
 
    y_pred = model.predict(X_test)
    output_metrics.log_metric("accuracy", accuracy_score(y_test, y_pred))
    output_metrics.log_metric("f1_score", f1_score(y_test, y_pred, average="weighted"))
 
    with open(output_model.path, "wb") as f:
        pickle.dump(model, f)
 
@dsl.pipeline(name="ml-training-pipeline")
def training_pipeline(
    data_path: str = "gs://ml-data/orders/v3/train.csv",
    n_estimators: int = 100
):
    ingest = data_ingestion(data_path=data_path)
    validate = data_validation(input_data=ingest.outputs["output_data"])
    train = train_model(
        input_data=ingest.outputs["output_data"],
        n_estimators=n_estimators
    )
    # 数据验证失败则跳过训练
    train.after(validate)