Posted on by Sharath Chandran

How to deploy machine learning models

How to deploy machine learning models ?

Machine learning models can be developed using TensorFlow, and orchastrated using Kubernetes and build it in Docker container. These can be combined with AWS services like CloudWatch, EKS and S3. This can basically give an API for developers to deploy their machine learning models.

What is Tensorflow?

Tensorflow is an opensource library used for machine learning specifically artificial neural networks. It provides API for various machine learning tasks.
It is important to learn following terminology prior learning Tensorflow.
Artificial neural networks are inspired from Animal brains that learn from experiences. Similarly neural networks learns from examples to perform its tasks than running predefined programs.


What is Keras

High level neural networks API written in python which Tensorflow utilized for building its programs.


What is Dense

Dense is the neural network layer in the tensorflow 

Model data drift monitoring using PCA 

import numpy as np
import matplotlib.pyplot as plt
from sklearn.decomposition import PCA
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error

=== 1. Load Iris and simulate drift ===

X = load_iris().data
X_train, X_test = train_test_split(X, test_size=0.5, random_state=42)

Inject synthetic drift into first 25 records in test set

X_test[:25, 0] += 2

=== 2. Scale Data ===

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

=== 3. PCA Train on Normal Data ===

pca = PCA(n_components=2)
X_train_pca = pca.fit_transform(X_train_scaled)
X_train_recon = pca.inverse_transform(X_train_pca)

=== 4. Establish Drift Threshold ===

mse_train = np.mean((X_train_scaled - X_train_recon) ** 2, axis=1)
threshold = np.percentile(mse_train, 95)
print(f"PCA Drift Detection Threshold: {threshold:.4f}")

=== 5. Apply to Incoming Test Data ===

errors = []
drift_flags = []

for i, x in enumerate(X_test_scaled):
x = x.reshape(1, -1)
x_pca = pca.transform(x)
x_recon = pca.inverse_transform(x_pca)
error = mean_squared_error(x, x_recon)
errors.append(error)
drift_flags.append(error > threshold)

=== 6. Plot the Drift Detection ===

plt.figure(figsize=(14, 6))
plt.style.use('seaborn-v0_8-poster')

Plot all errors

colors = ['#66c2a5' if not flag else '#fc8d62' for flag in drift_flags]
bars = plt.bar(range(len(errors)), errors, color=colors, edgecolor='black')

Highlight threshold

plt.axhline(y=threshold, color='purple', linestyle='--', linewidth=2, label=f"Threshold = {threshold:.4f}")

Gradient background

ax = plt.gca()
ax.set_facecolor('#f5f5f5')
ax.grid(True, linestyle='--', alpha=0.6)

Title and labels

plt.title("PCA Reconstruction Error for Drift Detection", fontsize=18, weight='bold')
plt.xlabel("Sample Index")
plt.ylabel("Reconstruction Error (MSE)")
plt.legend(loc='upper right')

Annotate drift points

for i, (err, is_drift) in enumerate(zip(errors, drift_flags)):
if is_drift:
plt.text(i, err + 0.02, "Drift", ha='center', va='bottom', fontsize=8, color='red', fontweight='bold')

plt.tight_layout()
plt.show()

model drift monitoring using Auto encoder

import numpy as np
import torch
import torch.nn as nn
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt

—————————

1. Define the Autoencoder

—————————

class Autoencoder(nn.Module):
def init(self, input_dim):
super().init()
self.encoder = nn.Sequential(
nn.Linear(input_dim, 16),
nn.ReLU(),
nn.Linear(16, 8)
)
self.decoder = nn.Sequential(
nn.Linear(8, 16),
nn.ReLU(),
nn.Linear(16, input_dim)
)

def forward(self, x):
    encoded = self.encoder(x)
    return self.decoder(encoded)

—————————

2. Simulate training data

—————————

np.random.seed(0)
X_train = np.random.normal(loc=0, scale=1.0, size=(1000, 10))

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_train_tensor = torch.tensor(X_train_scaled, dtype=torch.float32)

—————————

3. Train the Autoencoder

—————————

model = Autoencoder(input_dim=X_train.shape[1])
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

Train loop

model.train()
for epoch in range(100):
output = model(X_train_tensor)
loss = criterion(output, X_train_tensor)
optimizer.zero_grad()
loss.backward()
optimizer.step()

print(f”Training complete. Final loss: {loss.item():.4f}”)

—————————

4. Set a Drift Threshold

—————————

model.eval()
with torch.no_grad():
recon_train = model(X_train_tensor)
errors = torch.mean((X_train_tensor – recon_train) ** 2, dim=1)
threshold = np.percentile(errors.numpy(), 95) # 95th percentile
print(f”Drift threshold (95th percentile): {threshold:.4f}”)

—————————

5. Simulate live data feed

—————————

Simulate normal and drifting data

X_live_normal = np.random.normal(loc=0, scale=1.0, size=(100, 10))
X_live_drifted = np.random.normal(loc=2.0, scale=1.0, size=(100, 10))
X_live = np.vstack([X_live_normal, X_live_drifted])

Scale using training scaler

X_live_scaled = scaler.transform(X_live)
X_live_tensor = torch.tensor(X_live_scaled, dtype=torch.float32)

—————————

6. Monitor for Drift

—————————

drift_flags = []
errors_live = []

model.eval()
with torch.no_grad():
for i in range(X_live_tensor.shape[0]):
x = X_live_tensor[i].unsqueeze(0)
recon = model(x)
error = torch.mean((x – recon) ** 2).item()
errors_live.append(error)
is_drift = error > threshold
drift_flags.append(is_drift)
print(f”[{i}] Error: {error:.4f} {‘<– Drift’ if is_drift else ”}”)

—————————

7. Plot Results

—————————

plt.figure(figsize=(12, 4))
plt.plot(errors_live, label=”Reconstruction Error”)
plt.axhline(y=threshold, color=’r’, linestyle=’–‘, label=”Drift Threshold”)
plt.title(“Drift Detection with Autoencoder”)
plt.xlabel(“Data Point”)
plt.ylabel(“Reconstruction Error”)
plt.legend()
plt.show()

import numpy as np
import torch
import torch.nn as nn
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt

# ---------------------------
# 1. Define the Autoencoder
# ---------------------------
class Autoencoder(nn.Module):
    def __init__(self, input_dim):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 16),
            nn.ReLU(),
            nn.Linear(16, 8)
        )
        self.decoder = nn.Sequential(
            nn.Linear(8, 16),
            nn.ReLU(),
            nn.Linear(16, input_dim)
        )

    def forward(self, x):
        encoded = self.encoder(x)
        return self.decoder(encoded)

# ---------------------------
# 2. Simulate training data
# ---------------------------
np.random.seed(0)
X_train = np.random.normal(loc=0, scale=1.0, size=(1000, 10))

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_train_tensor = torch.tensor(X_train_scaled, dtype=torch.float32)

# ---------------------------
# 3. Train the Autoencoder
# ---------------------------
model = Autoencoder(input_dim=X_train.shape[1])
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

# Train loop
model.train()
for epoch in range(100):
    output = model(X_train_tensor)
    loss = criterion(output, X_train_tensor)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()

print(f"Training complete. Final loss: {loss.item():.4f}")

# ---------------------------
# 4. Set a Drift Threshold
# ---------------------------
model.eval()
with torch.no_grad():
    recon_train = model(X_train_tensor)
    errors = torch.mean((X_train_tensor - recon_train) ** 2, dim=1)
    threshold = np.percentile(errors.numpy(), 95)  # 95th percentile
print(f"Drift threshold (95th percentile): {threshold:.4f}")

# ---------------------------
# 5. Simulate live data feed
# ---------------------------
# Simulate normal and drifting data
X_live_normal = np.random.normal(loc=0, scale=1.0, size=(100, 10))
X_live_drifted = np.random.normal(loc=2.0, scale=1.0, size=(100, 10))
X_live = np.vstack([X_live_normal, X_live_drifted])

# Scale using training scaler
X_live_scaled = scaler.transform(X_live)
X_live_tensor = torch.tensor(X_live_scaled, dtype=torch.float32)

# ---------------------------
# 6. Monitor for Drift
# ---------------------------
drift_flags = []
errors_live = []

model.eval()
with torch.no_grad():
    for i in range(X_live_tensor.shape[0]):
        x = X_live_tensor[i].unsqueeze(0)
        recon = model(x)
        error = torch.mean((x - recon) ** 2).item()
        errors_live.append(error)
        is_drift = error > threshold
        drift_flags.append(is_drift)
        print(f"[{i}] Error: {error:.4f} {'<-- Drift' if is_drift else ''}")

# ---------------------------
# 7. Plot Results
# ---------------------------
plt.figure(figsize=(12, 4))
plt.plot(errors_live, label="Reconstruction Error")
plt.axhline(y=threshold, color='r', linestyle='--', label="Drift Threshold")
plt.title("Drift Detection with Autoencoder")
plt.xlabel("Data Point")
plt.ylabel("Reconstruction Error")
plt.legend()
plt.show()