MLOps

End-to-End ML Pipeline `` Data Source → Feature Store → Training Pipeline → Model Registry → Serving → Monitoring ↑ | Retraining trigger ←── Drift detection ←────────────┘ ``

MLOps Maturity Levels | Level | Description | |---|---| | 0 (Manual) | Ad-hoc scripts, manual steps, no tracking | | 1 (ML pipeline) | Automated training, experiment tracking, model registry | | 2 (CI/CD for ML) | Automated retraining, deployment, monitoring in CI/CD |

Core Components
• Experiment Tracking: MLflow, Weights & Biases, Comet
• Data Versioning: DVC, Delta Lake, LakeFS
• Pipeline Orchestration: Airflow, Kubeflow, Prefect, ZenML
• Feature Store: Feast, Tecton, Vertex Feature Store
• Model Registry: MLflow, BentoML, Seldon
• Serving: BentoML, TorchServe, TFServing, Ray Serve
• Monitoring: Evidently, Fiddler, Arize, WhyLabs

MLflow Tracking ```python import mlflow import mlflow.sklearn from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import accuracy_score, f1_score

mlflow.set_experiment("churn-prediction-v2")

with mlflow.start_run(run_name="rf-tuned"): # Log hyperparameters params = {"n_estimators": 200, "max_depth": 8, "min_samples_leaf": 10} mlflow.log_params(params)

# Train model = RandomForestClassifier(**params) model.fit(X_train, y_train) y_pred = model.predict(X_test)

# Log metrics mlflow.log_metrics({ "accuracy": accuracy_score(y_test, y_pred), "f1": f1_score(y_test, y_pred, average="weighted"), "train_size": len(X_train) })

# Log model to registry mlflow.sklearn.log_model( model, "model", registered_model_name="ChurnPredictor", input_example=X_train[:5] )

# Log artifacts mlflow.log_artifact("feature_importance.png") mlflow.log_artifact("confusion_matrix.png") ```

Weights & Biases ```python import wandb

wandb.init(project="churn-prediction", config={ "n_estimators": 200, "max_depth": 8, "learning_rate": 0.01 })

for epoch in range(100): loss = train_epoch() wandb.log({"loss": loss, "epoch": epoch})

wandb.finish() ```

BentoML Serving ```python import bentoml from bentoml.io import JSON, NumpyNdarray

# Save model to BentoML bentoml.sklearn.save_model("churn_model", model)

# Define service svc = bentoml.Service("churn_predictor")

@svc.api(input=JSON(), output=JSON()) async def predict(input_data: dict): runner = bentoml.sklearn.get("churn_model:latest").to_runner() features = preprocess(input_data) prediction = await runner.predict.async_run(features) return {"churn_probability": float(prediction[0]), "threshold": 0.5}

# serve: bentoml serve service:svc --port 3000 # Build docker: bentoml build && bentoml containerize ```

FastAPI Serving (simple) ```python from fastapi import FastAPI import joblib import numpy as np

app = FastAPI() model = joblib.load("model.pkl") scaler = joblib.load("scaler.pkl")

@app.post("/predict") async def predict(data: dict): features = np.array(list(data.values())).reshape(1, -1) features_scaled = scaler.transform(features) prob = model.predict_proba(features_scaled)[0][1] return {"churn_probability": prob, "prediction": prob > 0.5} ```

Blue/Green ML Deployment ``python # Traffic splitting between model versions router = { "model_v1": {"endpoint": "v1-service:8000", "weight": 0.2}, "model_v2": {"endpoint": "v2-service:8000", "weight": 0.8} } # Monitor: if v2 error rate increases, shift traffic back to v1 ``

Data Drift Detection ```python from evidently import ColumnMapping from evidently.report import Report from evidently.metric_preset import DataDriftPreset, DataQualityPreset

column_mapping = ColumnMapping( target="churn", numerical_features=["age", "tenure", "monthly_charges"], categorical_features=["contract_type", "internet_service"] )

report = Report(metrics=[DataDriftPreset(), DataQualityPreset()]) report.run(reference_data=train_df, current_data=production_df, column_mapping=column_mapping) report.save_html("drift_report.html") ```

Prediction Monitoring ```python def monitor_predictions(model, features, ground_truth=None): predictions = model.predict_proba(features)

# Distribution check (is prediction dist shifting?) from scipy.stats import ks_2samp stat, p_value = ks_2samp(reference_preds, predictions[:, 1])

if p_value < 0.05: alert("Prediction distribution drift detected!", severity="warning")

# Accuracy check (if labels available with delay) if ground_truth is not None: current_accuracy = accuracy_score(ground_truth, predictions.argmax(1)) if current_accuracy < threshold: alert(f"Accuracy dropped to {current_accuracy:.3f}", severity="critical") trigger_retraining() ```

Automated Retraining Trigger ```python # Airflow DAG for weekly retraining from airflow import DAG from airflow.operators.python import PythonOperator

with DAG("model_retraining", schedule="@weekly") as dag: check_drift = PythonOperator(task_id="check_drift", python_callable=run_drift_check) retrain = PythonOperator(task_id="retrain", python_callable=train_model) evaluate = PythonOperator(task_id="evaluate", python_callable=evaluate_new_model) deploy = PythonOperator(task_id="deploy", python_callable=deploy_if_better)

check_drift >> retrain >> evaluate >> deploy ```

Why Feature Stores Features are the most expensive part of ML. Without a feature store, every team recomputes the same features independently, creating inconsistency between training and serving (training-serving skew).

from feast import FeatureStore, Entity, FeatureView, Field
from feast.types import Float32, Int64

# ── Define feature store ────────────────────────────── store = FeatureStore(repo_path=".")

# Entity user = Entity( name="user_id", description="User identifier" )

# Feature view from offline source from feast import FileSource from datetime import timedelta

user_features_source = FileSource( path="data/user_features.parquet", timestamp_field="event_timestamp" )

user_feature_view = FeatureView( name="user_features", entities=[user], ttl=timedelta(days=7), schema=[ Field(name="purchase_count_30d", dtype=Int64), Field(name="avg_order_value", dtype=Float32), Field(name="days_since_last_buy", dtype=Int64), Field(name="churn_risk_score", dtype=Float32), ], source=user_features_source, )

# ── Training: fetch historical features ─────────────── entity_df = pd.DataFrame({ "user_id": [1, 2, 3, 4, 5], "event_timestamp": pd.to_datetime(["2024-01-15"] * 5), })

training_df = store.get_historical_features( entity_df=entity_df, features=["user_features:purchase_count_30d", "user_features:avg_order_value", "user_features:churn_risk_score"] ).to_df()

# ── Serving: fetch real-time features ──────────────── feature_vector = store.get_online_features( features=["user_features:purchase_count_30d", "user_features:churn_risk_score"], entity_rows=[{"user_id": 42}] ).to_dict() # Guaranteed same feature computation as training ```