Common ML Service Misuses And Prevention Strategies

In the realm of Machine Learning (ML), deploying models effectively requires more than just building them. It involves ensuring their reliability, performance, and maintainability over time. This article dives into some prevalent ML service misuses that can impact your projects and provides actionable strategies to prevent them. We'll explore issues identified in a project discussion, offering insights and refactored code examples to help you build robust and dependable ML systems.

🔍 Detected ML Service Misuses

When working with Machine Learning services, several pitfalls can compromise your project's success. Let's examine some common misuses and how to address them.

Ignoring Monitoring Data Drift

One critical aspect of maintaining ML model performance is monitoring for data drift. Data drift occurs when the statistical properties of the input data change over time, leading to a degradation in model accuracy. Ignoring this phenomenon can result in inaccurate predictions and unreliable results. It's crucial to establish mechanisms for detecting both input and output drift. One way to think about it is by envisioning a bridge that was structurally sound when built but experiences degradation from a recent earthquake. Ignoring that degradation makes the bridge unstable. Similarly, ignoring data drift makes your model’s output unreliable.

Why Monitoring Data Drift is Essential

• Maintaining Accuracy: Models trained on historical data may become less accurate as new data deviates from the original distribution.
• Ensuring Reliability: Consistent performance is vital for applications relying on ML predictions. Data drift can undermine this reliability.
• Improving Model Maintainability: Detecting drift early allows for timely retraining or model adjustments, simplifying maintenance.

Strategies for Detecting Data Drift

• Statistical Tests: Employ statistical tests like the Kolmogorov-Smirnov test or the Chi-squared test to compare the distributions of input features over time.
• Drift Detection Algorithms: Utilize specialized algorithms designed to identify drift, such as the Population Stability Index (PSI) or the concept drift detection methods.
• Visualizations: Regularly plot data distributions to visually inspect for shifts or changes.

Refactored Code Example

The provided code snippet lacks a mechanism to detect input/output drift across time. Here’s a refactored version incorporating basic drift monitoring:

import time
from datetime import datetime
import pandas as pd

# Load libraries
import torch
import torch.nn as nn
import torch.optim as optim
from scipy.stats import ks_2samp

# Define constants
MAX_EPOCHS = 10
BATCH_SIZE = 32

class MyModel(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim):
        super(MyModel, self).__init__()
        self.fc1 = nn.Linear(input_dim, hidden_dim)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(hidden_dim, output_dim)

    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        return x

# Load data
train_data = pd.read_csv('train.csv')
test_data = pd.read_csv('test.csv')

# Split data into training and validation sets
train_size = int(0.8 * len(train_data))
train_data, val_data = train_data[:train_size], train_data[train_size:]

# Define model and optimizer
model = MyModel(input_dim=784, hidden_dim=256, output_dim=10)
optimizer = optim.SGD(model.parameters(), lr=0.01)

# Function to detect data drift using Kolmogorov-Smirnov test
def detect_data_drift(reference_data, current_data, threshold=0.05):
    drift_detected = False
    for column in reference_data.columns:
        ks_statistic, p_value = ks_2samp(reference_data[column], current_data[column])
        if p_value < threshold:
            print(f'Data drift detected in column {column}: p-value = {p_value}')
            drift_detected = True
    return drift_detected

# Store a reference batch of data for drift detection
reference_data = train_data.sample(frac=0.1)

# Train model
for epoch in range(MAX_EPOCHS):
    for x, y in zip(train_data.drop('target', axis=1).values, train_data['target'].values):
        # Forward pass
        outputs = model(torch.tensor(x, dtype=torch.float))
        loss = nn.CrossEntropyLoss()(outputs, torch.tensor(y))

        # Backward and optimize
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

    # Print loss at each epoch
    print(f'Epoch {epoch+1}, Loss: {loss.item()}')

    # Simulate new data
    new_data = pd.read_csv('test.csv').sample(frac=0.1)

    # Detect data drift
    if detect_data_drift(reference_data, new_data.drop('target', axis=1)):
        print('Data drift detected, consider retraining the model')
    else:
        print('No significant data drift detected')

# Evaluate model on test data
test_loss = 0
correct = 0
with torch.no_grad():
    for x, y in zip(test_data.drop('target', axis=1).values, test_data['target'].values):
        outputs = model(torch.tensor(x, dtype=torch.float))
        loss = nn.CrossEntropyLoss()(outputs, torch.tensor(y))
        test_loss += loss.item()
        _, predicted = torch.max(outputs, 1)
        correct += (predicted == torch.tensor(y)).sum().item()

print(f'Test Loss: {test_loss / len(test_data)}')
print(f'Test Accuracy: {correct / len(test_data)}')

This refactored code incorporates a detect_data_drift function using the Kolmogorov-Smirnov test. This test helps determine if two datasets come from the same distribution. By comparing a reference batch of data with new data, you can identify significant deviations and take appropriate action, such as retraining the model.

Ignoring Testing Schema Mismatch

Another common pitfall is invoking a model without validating the input schema or data type compatibility. This schema mismatch can lead to unexpected errors and unreliable predictions. It's essential to ensure that the data being fed into the model aligns with the expected format and types. Failing to test for schema mismatch is like trying to fit a square peg in a round hole – it simply won’t work and can cause damage.

Why Testing Schema Mismatch is Critical

• Preventing Runtime Errors: Validating input data can prevent errors that arise from incompatible data structures.
• Ensuring Data Integrity: Schema validation helps maintain the integrity of the data processed by the model.
• Improving Model Stability: Consistent data input leads to more stable and predictable model behavior.

Strategies for Preventing Schema Mismatch

• Schema Validation Libraries: Use libraries like Cerberus or Marshmallow to define and enforce data schemas.
• Type Checking: Implement type checking mechanisms to ensure data types match the model's expectations.
• Data Contracts: Establish clear data contracts that outline the expected structure and types of input data.

Refactored Code Example

The original code lacks input schema validation. Here’s a refactored version incorporating basic schema validation:

import time
from datetime import datetime
import jsonschema
from jsonschema import validate
import pandas as pd
import torch
import torch.nn as nn
import torch.optim as optim


# Define input schema
input_schema = {
    "type": "array",
    "items": {
        "type": "number"
    },
    "minItems": 784,
    "maxItems": 784
}

def validate_input(data, schema):
    try:
        validate(instance=data, schema=schema)
        return True
    except jsonschema.exceptions.ValidationError as e:
        print(f"Input validation error: {e}")
        return False


def train_cl(model, train_datasets, replay_mode="", scenario="", classes_per_task=0, iters=10000, batch_size=32):
    # your training logic here
    pass

class AutoEncoder(nn.Module):
    def __init__(self, image_size, image_channels, fc_layers, fc_units, z_dim, classes, fc_drop, fc_bn, fc_nl):
        super(AutoEncoder, self).__init__()
        self.fc1 = nn.Linear(image_size * image_size * image_channels, fc_units)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(fc_units, z_dim)

    def forward(self, x):
        x = self.fc1(x)
        x = self.relu()
        x = self.fc2(x)
        return x

# original main function
if __name__ == "__main__":
    start = time.time()
    param_stamp = datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
    
    # Example configuration
    config = {"size": 28, "channels": 1, "classes": 10}
    
    # Example command-line arguments
    class Args:
        def __init__(self):
            self.fc_lay = 2
            self.fc_uni = 256
            self.z_dim = 32
            self.fc_drop = 0.1
            self.fc_bn = True
            self.fc_nl = "relu"
            self.lr = 0.001
            self.tasks = 10
    
    args = Args()
    
    model = AutoEncoder(
        image_size=config['size'], image_channels=config['channels'],
        fc_layers=args.fc_lay, fc_units=args.fc_uni,
        z_dim=args.z_dim, classes=config['classes'],
        fc_drop=args.fc_drop, fc_bn=True if args.fc_bn else False,
        fc_nl=args.fc_nl
    )
    model.optim_list = [{'params': filter(lambda p: p.requires_grad, model.parameters()), 'lr': args.lr}]
    model.optim_type = "adam"
    if model.optim_type == "adam":
        model.optimizer = optim.Adam(model.optim_list, betas=(0.9, 0.999))
    elif model.optim_type == "sgd":
        model.optimizer = optim.SGD(model.optim_list)

    iters = 10000  # Define iters here
    for i in range(args.tasks):
        for j in range(iters):
            # Example input data
            input_data = list(pd.read_csv('train.csv').sample(n=1).drop('target', axis=1).values.flatten())
            # Validate input data against the schema
            if validate_input(input_data, input_schema):
                # Your training logic here
                pass
            else:
                print("Skipping training step due to invalid input data")

    end = time.time()
    print(f"Training took {end - start} seconds")

In this example, a JSON schema is defined to represent the expected input format. The validate_input function then checks incoming data against this schema, ensuring that it conforms to the required structure before being processed by the model. This proactive validation step helps prevent runtime errors and ensures data integrity.

Not Using Training Checkpoints

A critical yet often overlooked aspect of ML model training is the implementation of checkpoints. Without checkpoints, you risk losing progress if your training process is interrupted due to crashes, power outages, or other unforeseen issues. Training checkpoints act as snapshots of your model's state at various intervals, allowing you to resume training from the last saved state rather than starting from scratch. Neglecting training checkpoints is like building a house without a foundation – it may look good initially, but it’s vulnerable to collapse.

Why Training Checkpoints are Essential

• Preventing Data Loss: Checkpoints safeguard your training progress, minimizing the impact of interruptions.
• Enabling Experimentation: You can revert to previous checkpoints to explore different training paths or hyperparameters.
• Improving Efficiency: Resuming from a checkpoint saves valuable time and resources by avoiding redundant training.

Strategies for Implementing Training Checkpoints

• Periodic Saving: Save model checkpoints at regular intervals, such as after each epoch or a set number of iterations.
• Performance-Based Saving: Save checkpoints when the model achieves a new performance milestone, such as a higher validation accuracy.
• Cloud-Based Storage: Store checkpoints in cloud storage to ensure durability and accessibility.

Refactored Code Example

The original code lacks training checkpoint functionality. Here’s a refactored version incorporating checkpointing using the sagemaker library:

%matplotlib inline
import sys
from urllib.request import urlretrieve
import zipfile
from dateutil.parser import parse
import json
from random import shuffle
import random
import datetime
import os
import boto3
import s3fs
import sagemaker
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from __future__ import print_function
from ipywidgets import interact, interactive, fixed, interact_manual
import ipywidgets as widgets
from ipywidgets import IntSlider, FloatSlider, Checkbox
np.random.seed(1)
!conda install -y s3fs
import boto3
import s3fs
import sagemaker
from sagemaker import get_execution_role

# Input your own file path in the below lines
bucket='{{{bucket_name}}}'
prefix = 'prysmian-forecasting'
sagemaker_session = sagemaker.Session()
role = get_execution_role()
folder1='sagemaker'
folder2='data'
folder3='train'
data_key = 'RawMaterialItems.xlsx'
s3_data_path = 's3://{}/{}/{}/{}/{}'.format(bucket, folder1, folder2, folder3, data_key)
s3_output_path = "s3://{}/{}/output".format(bucket, prefix)

from sagemaker.amazon.amazon_estimator import get_image_uri
image_name = get_image_uri(boto3.Session().region_name, 'forecasting-deepar')
freq = 'M'
prediction_length = 5
context_length = 12
data = pd.read_excel(s3_data_path, parse_dates=True, index_col=0)
num_timeseries = data.shape[1]
data_length=data.index.size
print("This is the number of time series you're running through the algorithm (This many materials):")
print(num_timeseries)
print("This is the number of data points for each time series:")
print(data_length)
t0 = data.index[0]
print("This is the beginning date:")
print(t0)
time_series=[]
for i in range(num_timeseries):
    index = pd.DatetimeIndex(start=t0, freq=freq, periods=data_length)
    time_series.append(pd.Series(data=data.iloc[:,i], index=index))
print(time_series[2])
time_series[2].plot()
plt.show()
time_series_training = []
for ts in time_series:
    time_series_training.append(ts[:-prediction_length])
time_series[2].plot(label='test')
time_series_training[2].plot(label='train', ls=':')
plt.legend()
plt.show()

def series_to_obj(ts, cat=None):
    obj = {"start": str(ts.index[0]), "target": list(ts)}
    if cat is not None:
        obj["cat"] = cat
    return obj

def series_to_jsonline(ts, cat=None):
    return json.dumps(series_to_obj(ts, cat))

encoding = "utf-8"
s3filesystem = s3fs.S3FileSystem()
with s3filesystem.open(s3_data_path + "/train/train.json", 'wb') as fp:
    for ts in time_series_training:
        fp.write(series_to_jsonline(ts).encode(encoding))
        fp.write('\n'.encode(encoding))
with s3filesystem.open(s3_data_path + "/test/test.json", 'wb') as fp:
    for ts in time_series:
        fp.write(series_to_jsonline(ts).encode(encoding))
        fp.write('\n'.encode(encoding))

estimator = sagemaker.estimator.Estimator(
    sagemaker_session=sagemaker_session,
    image_name=image_name,
    role=role,
    train_instance_count=1,
    train_instance_type='ml.c4.xlarge',
    base_job_name='DEMO-deepar',
    output_path=s3_output_path,
    # Added checkpointing configuration
    checkpoint_s3_uri=f"s3://{bucket}/{prefix}/checkpoints",
    checkpoint_local_path='/opt/ml/checkpoints'
)

hyperparameters = {
    "time_freq": freq,
    "context_length": str(context_length),
    "prediction_length": str(prediction_length),
    "num_cells": "40",
    "num_layers": "3",
    "likelihood": "gaussian",
    "epochs": "80",
    "mini_batch_size": "32",
    "learning_rate": "0.001",
    "dropout_rate": "0.05",
    "early_stopping_patience": "10"
}
estimator.set_hyperparameters(**hyperparameters)

# Ensure the S3 path exists before training
s3_checkpoints_path = f"s3://{bucket}/{prefix}/checkpoints"

# Try creating the directory, it's okay if it already exists
try:
    boto3.resource('s3').Object(bucket, f'{prefix}/checkpoints/').put(Body=b'')
    print(f"Created S3 path: {s3_checkpoints_path}")
except Exception as e:
    print(f"Error creating S3 path (may already exist): {e}")

# Train the model with checkpointing
estimator.fit(inputs={"train": s3_data_path + "/train/", "test": s3_data_path + "/test/"}, wait=True)

job_name = estimator.latest_training_job.name
endpoint_name = sagemaker_session.endpoint_from_job(
    job_name=job_name,
    initial_instance_count=1,
    instance_type='ml.m4.xlarge',
    deployment_image=image_name,
    role=role
)

class DeepARPredictor(sagemaker.predictor.RealTimePredictor):
    def set_prediction_parameters(self, freq, prediction_length):
        """Set the time frequency and prediction length parameters. This method **must** be called
        before being able to use `predict`.
        Parameters:
        freq -- string indicating the time frequency
        prediction_length -- integer, number of predicted time points
        Return value: none.
        """
        self.freq = freq
        self.prediction_length = prediction_length

    def predict(self, ts, cat=None, encoding="utf-8", num_samples=100, quantiles=["0.1", "0.75", "0.9"]):
        """Requests the prediction of for the time series listed in `ts`, each with the (optional)
        corresponding category listed in `cat`.
        Parameters:
        ts -- list of `pandas.Series` objects, the time series to predict
        cat -- list of integers (default: None)
        encoding -- string, encoding to use for the request (default: "utf-8")
        num_samples -- integer, number of samples to compute at prediction time (default: 100)
        quantiles -- list of strings specifying the quantiles to compute (default: ["0.1", "0.5", "0.9"])
        Return value: list of `pandas.DataFrame` objects, each containing the predictions
        """
        prediction_times = [x.index[-1]+1 for x in ts]
        req = self.__encode_request(ts, cat, encoding, num_samples, quantiles)
        res = super(DeepARPredictor, self).predict(req)
        return self.__decode_response(res, prediction_times, encoding)

    def __encode_request(self, ts, cat, encoding, num_samples, quantiles):
        instances = [series_to_obj(ts[k], cat[k] if cat else None) for k in range(len(ts))]
        configuration = {"num_samples": num_samples, "output_types": ["quantiles"], "quantiles": quantiles}
        http_request_data = {"instances": instances, "configuration": configuration}
        return json.dumps(http_request_data).encode(encoding)

    def __decode_response(self, response, prediction_times, encoding):
        response_data = json.loads(response.decode(encoding))
        list_of_df = []
        for k in range(len(prediction_times)):
            prediction_index = pd.DatetimeIndex(start=prediction_times[k], freq=self.freq, periods=self.prediction_length)
            list_of_df.append(pd.DataFrame(data=response_data['predictions'][k]['quantiles'], index=prediction_index))
        return list_of_df

predictor = DeepARPredictor(
    endpoint=endpoint_name,
    sagemaker_session=sagemaker_session,
    content_type="application/json"
)

predictor.set_prediction_parameters(freq, prediction_length)
list_of_df = predictor.predict(time_series_training[:60])
actual_data = time_series[:5]

for k in range(len(list_of_df)):
    plt.figure(figsize=(12,6))
    actual_data[k][-prediction_length-context_length:].plot(label='target')
    p10 = list_of_df[k]['0.1']
    p90 = list_of_df[k]['0.9']
    plt.fill_between(p10.index, p10, p90, color='y', alpha=0.5, label='80% confidence interval')
    list_of_df[k]['0.75'].plot(label='prediction median')
    plt.legend()
    plt.show()

print(predictor.predict(time_series[:4]))
sagemaker_session.delete_endpoint(endpoint_name)

Key changes in this refactored code:

• Checkpoint Configuration: The checkpoint_s3_uri and checkpoint_local_path parameters are added to the sagemaker.estimator.Estimator constructor, specifying where to store checkpoints in S3 and locally, respectively.
• S3 Path Creation: Code is added to ensure the S3 path for checkpoints exists before training begins.
• Training with Checkpointing: The estimator.fit() method is called to train the model with checkpointing enabled. SageMaker automatically saves checkpoints during training, which can be used to resume training if needed.

By implementing these changes, your model training process becomes more resilient and efficient, safeguarding against data loss and enabling easier experimentation.

📝 Conclusion

Avoiding common ML service misuses is crucial for building robust, reliable, and maintainable Machine Learning systems. By monitoring for data drift, validating input schemas, and utilizing training checkpoints, you can significantly improve the performance and stability of your models. The refactored code examples provided offer practical guidance on how to implement these preventative measures in your projects.

For more in-depth information on best practices in Machine Learning, consider exploring resources like the MLOps Guide. This will help you gain a deeper understanding of how to build and deploy ML models effectively.