ML Deployment: Productionizing Machine Learning Models

# batch_scoring.py
import joblib
import pandas as pd
from sqlalchemy import create_engine

def load_model(model_path):
    return joblib.load(model_path)

def batch_predict(model, input_data):
    predictions = model.predict(input_data)
    probabilities = model.predict_proba(input_data)
    return {
        'predictions': predictions.tolist(),
        'probabilities': probabilities.tolist()
    }

def save_results(results, table_name):
    engine = create_engine('postgresql://user:pass@localhost/db')
    df = pd.DataFrame(results)
    df.to_sql(table_name, engine, if_exists='replace')

if __name__ == '__main__':
    model = load_model('models/classifier.pkl')
    
    # Load data from database
    engine = create_engine('postgresql://user:pass@localhost/db')
    data = pd.read_sql('SELECT * FROM features_table', engine)
    
    # Run predictions
    results = batch_predict(model, data)
    
    # Save results
    save_results(results, 'predictions')

# api.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI()
model = joblib.load('models/classifier.pkl')

class PredictionRequest(BaseModel):
    feature1: float
    feature2: float
    feature3: float

class PredictionResponse(BaseModel):
    prediction: int
    probability: float

@app.post('/predict', response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    features = np.array([[
        request.feature1,
        request.feature2,
        request.feature3
    ]])
    
    prediction = model.predict(features)[0]
    probability = model.predict_proba(features)[0].max()
    
    return PredictionResponse(
        prediction=int(prediction),
        probability=float(probability)
    )

@app.get('/health')
async def health():
    return {'status': 'healthy'}

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY models ./models
COPY api.py .

EXPOSE 8000

CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ml-model
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ml-model
  template:
    metadata:
      labels:
        app: ml-model
    spec:
      containers:
      - name: model
        image: myregistry/ml-model:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
---
apiVersion: v1
kind: Service
metadata:
  name: ml-model-service
spec:
  selector:
    app: ml-model
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

# monitoring.py
from prometheus_client import Counter, Histogram, Gauge

prediction_counter = Counter('predictions_total', 'Total predictions')
prediction_latency = Histogram('prediction_latency_seconds', 'Prediction latency')
model_accuracy = Gauge('model_accuracy', 'Model accuracy')

def track_prediction(features):
    with prediction_latency.time():
        prediction = model.predict(features)
        prediction_counter.inc()
        return prediction

def update_accuracy(y_true, y_pred):
    accuracy = accuracy_score(y_true, y_pred)
    model_accuracy.set(accuracy)

# drift_detection.py
from scipy import stats
import numpy as np

def detect_drift(reference_data, current_data, threshold=0.05):
    """
    Detect if current data distribution differs from reference
    """
    drift_detected = False
    drift_report = {}
    
    for feature in reference_data.columns:
        ref_values = reference_data[feature].values
        curr_values = current_data[feature].values
        
        # Kolmogorov-Smirnov test
        statistic, p_value = stats.ks_2samp(ref_values, curr_values)
        
        if p_value < threshold:
            drift_detected = True
            drift_report[feature] = {
                'statistic': statistic,
                'p_value': p_value,
                'drift': True
            }
        else:
            drift_report[feature] = {'drift': False}
    
    return drift_detected, drift_report