Production ML Pipelines Beyond Analytics Automation

from zenml import pipeline, step, Model
from zenml.integrations.mlflow.steps import (
    mlflow_model_deployer_step,
)
import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_squared_error
import numpy as np

@step
def ingest_data() -> pd.DataFrame:
    return pd.read_csv("data/dataset.csv")

@step
def train_model(df: pd.DataFrame) -> RandomForestRegressor:
    X, y = df.drop("target", axis=1), df["target"]
    model = RandomForestRegressor(n_estimators=100)
    model.fit(X, y)
    return model

@step
def evaluate(model: RandomForestRegressor, df: pd.DataFrame) -> float:
    X, y = df.drop("target", axis=1), df["target"]
    preds = model.predict(X)
    return float(np.sqrt(mean_squared_error(y, preds)))

@step
def check_drift(df: pd.DataFrame) -> bool:
    # Plug in Evidently, Great Expectations, etc.
    return detect_drift(df)

@pipeline(model=Model(name="my_model"))
def ml_pipeline():
    df = ingest_data()
    model = train_model(df)
    rmse = evaluate(model, df)
    drift = check_drift(df)

# Runs on any orchestrator, logs to MLflow,
# tracks artifacts, and triggers retraining — all
# in one portable, version-controlled pipeline
ml_pipeline()