Machine learning pipelines are production systems with reliability requirements. This seems obvious when stated, but the organizational reality in most companies is that ML pipelines are owned by data scientists and ML engineers whose primary expertise is model performance, not system reliability. When the training pipeline fails at 3am, nobody knows whose problem it is. When model quality silently degrades, nobody has an alert for it. When the feature store goes stale, the production model is making predictions on data that no longer reflects reality.
SRE practices applied to ML pipelines close these gaps. Here's what ML pipeline reliability looks like in practice.
The ML Pipeline Reliability Stack
An ML production system has multiple components, each with distinct failure modes:
Data Sources (databases, event streams, APIs)
↓
Feature Engineering Pipeline (Spark/Flink/dbt)
↓
Feature Store (online + offline)
↓
Training Pipeline (SageMaker, Vertex AI, Databricks)
↓
Model Registry (MLflow, SageMaker Model Registry)
↓
Model Serving (inference endpoints, batch scoring)
↓
Monitoring (data drift, model drift, prediction quality)Reliability must be designed and measured at every layer. A failure at any layer degrades the whole system — and the failures don't always look like outages. They often look like "the model worked, but not correctly."
Feature Pipeline Reliability: Freshness and Completeness
The feature engineering pipeline produces the inputs that models use for both training and inference. Its reliability requirements are freshness (how current are the features?) and completeness (are all expected features present for all entities?).
Freshness SLO definition:
# Feature freshness monitoring
class FeatureFreshnessSLO:
def __init__(self, feature_group: str, max_staleness_minutes: int):
self.feature_group = feature_group
self.max_staleness = max_staleness_minutes
def check(self) -> SLOStatus:
latest_update = get_feature_store_last_updated(self.feature_group)
staleness = (datetime.utcnow() - latest_update).total_seconds() / 60
if staleness > self.max_staleness:
return SLOStatus(
compliant=False,
staleness_minutes=staleness,
message=f"Feature group {self.feature_group} is {staleness:.1f} minutes stale "
f"(max: {self.max_staleness} minutes)"
)
return SLOStatus(compliant=True, staleness_minutes=staleness)
# Define SLOs per feature group based on business requirements
slos = [
FeatureFreshnessSLO("user_transaction_features", max_staleness_minutes=60),
FeatureFreshnessSLO("product_inventory_features", max_staleness_minutes=5),
FeatureFreshnessSLO("fraud_risk_features", max_staleness_minutes=15),
]Completeness monitoring:
def check_feature_completeness(feature_group: str, entity_set: set) -> CompletenessReport:
"""
Verify that all expected entities have features in the store.
Missing features for active entities indicate a pipeline failure.
"""
stored_entities = get_entities_in_feature_store(feature_group)
missing = entity_set - stored_entities
coverage = len(stored_entities) / len(entity_set)
return CompletenessReport(
feature_group=feature_group,
expected_entities=len(entity_set),
stored_entities=len(stored_entities),
missing_entities=len(missing),
coverage_pct=coverage * 100,
compliant=coverage > 0.999 # SLO: 99.9% of entities must have features
)Data Drift Detection: The Silent Model Killer
A model trained on historical data makes predictions based on patterns it learned. When the real-world data distribution shifts — seasonally, due to business changes, or due to upstream data pipeline changes — the model's predictions may become increasingly wrong without any infrastructure failure occurring.
This is data drift, and it's the most common cause of silent ML reliability failures.
Input feature drift monitoring:
from scipy import stats
import numpy as np
class FeatureDriftMonitor:
def __init__(self, reference_distribution: dict, threshold: float = 0.05):
"""
reference_distribution: {feature_name: array of reference values}
threshold: p-value below which we flag as drifted
"""
self.reference = reference_distribution
self.threshold = threshold
def check_drift(self, current_distribution: dict) -> DriftReport:
drifted_features = []
for feature_name, reference_values in self.reference.items():
if feature_name not in current_distribution:
drifted_features.append({
"feature": feature_name,
"reason": "missing from current data"
})
continue
current_values = current_distribution[feature_name]
# Kolmogorov-Smirnov test: are these from the same distribution?
statistic, p_value = stats.ks_2samp(reference_values, current_values)
if p_value < self.threshold:
drifted_features.append({
"feature": feature_name,
"ks_statistic": statistic,
"p_value": p_value,
"reference_mean": np.mean(reference_values),
"current_mean": np.mean(current_values),
"delta_pct": (np.mean(current_values) - np.mean(reference_values))
/ np.mean(reference_values) * 100
})
return DriftReport(
drifted_features=drifted_features,
total_features=len(self.reference),
drift_detected=len(drifted_features) > 0
)Run drift detection on every batch of inference requests (or on a sample for high-volume online serving). Alert when drift is detected in features that the model relies heavily on. Drift in important features doesn't automatically mean the model is wrong — but it means you should validate model performance against recent outcomes.
Prediction distribution monitoring:
Track the distribution of your model's output (prediction scores, class probabilities) over time. If a fraud detection model that usually outputs scores averaging 0.15 suddenly starts averaging 0.05, something changed — either the input data shifted, or the model is responding differently. Neither is necessarily wrong, but both warrant investigation.
Training Pipeline Reliability
Training pipelines are batch processes with SLAs: "the weekly retrained model must be in the registry and validated by Sunday 6am for Monday's serving deployment."
Checkpoint-based training for resilience:
class CheckpointedTrainer:
def __init__(self, checkpoint_dir: str, s3_bucket: str):
self.checkpoint_dir = checkpoint_dir
self.s3_bucket = s3_bucket
def train(self, model, dataset, epochs: int):
start_epoch = self.load_latest_checkpoint(model)
for epoch in range(start_epoch, epochs):
self.train_epoch(model, dataset, epoch)
# Checkpoint every 5 epochs and on completion
if epoch % 5 == 0 or epoch == epochs - 1:
checkpoint_path = self.save_checkpoint(model, epoch)
self.upload_to_s3(checkpoint_path)
# Log checkpoint for observability
logger.info(json.dumps({
"event": "checkpoint_saved",
"epoch": epoch,
"checkpoint_path": checkpoint_path,
"model_loss": self.get_current_loss(model)
}))
def load_latest_checkpoint(self, model) -> int:
"""Resume from latest S3 checkpoint if available."""
latest = self.find_latest_s3_checkpoint()
if latest:
model.load_state_dict(torch.load(latest))
epoch = self.extract_epoch_from_path(latest)
logger.info(f"Resuming training from epoch {epoch}")
return epoch + 1
return 0Training job alerting:
# SageMaker training job monitoring
def monitor_training_job(job_name: str, expected_duration_hours: float):
sm = boto3.client('sagemaker')
while True:
response = sm.describe_training_job(TrainingJobName=job_name)
status = response['TrainingJobStatus']
if status == 'Completed':
log_training_completion(job_name, response)
return
elif status == 'Failed':
alert_training_failure(job_name, response['FailureReason'])
raise TrainingFailedError(response['FailureReason'])
# Check for jobs running longer than expected
elapsed_hours = (datetime.utcnow() - response['TrainingStartTime'].replace(tzinfo=None)).total_seconds() / 3600
if elapsed_hours > expected_duration_hours * 1.5:
alert_training_overtime(job_name, elapsed_hours, expected_duration_hours)
time.sleep(60)Model Registry as a Reliability Control
The model registry is the gating mechanism between a trained model and production. Before a model version can be promoted to production serving, it must pass evaluation gates:
class ModelPromotionGate:
def __init__(self, registry_client, thresholds: dict):
self.registry = registry_client
self.thresholds = thresholds
def evaluate(self, model_version: str, eval_dataset: str) -> PromotionDecision:
metrics = self.run_evaluation(model_version, eval_dataset)
failures = []
for metric, threshold in self.thresholds.items():
if metrics[metric] < threshold:
failures.append({
"metric": metric,
"actual": metrics[metric],
"required": threshold
})
if failures:
self.registry.update_model_version_stage(
model_version, stage="Rejected",
description=f"Failed gates: {failures}"
)
return PromotionDecision(approved=False, failures=failures)
# Compare to current production model
production_metrics = self.get_production_model_metrics()
regression = []
for metric, value in metrics.items():
production_value = production_metrics.get(metric)
if production_value and value < production_value * 0.95: # 5% regression threshold
regression.append({
"metric": metric,
"candidate": value,
"production": production_value,
"regression_pct": (production_value - value) / production_value * 100
})
if regression:
return PromotionDecision(approved=False, reason="Regression vs production", details=regression)
self.registry.update_model_version_stage(model_version, stage="Production")
return PromotionDecision(approved=True, metrics=metrics)
# Required gates for fraud detection model
fraud_gates = ModelPromotionGate(
registry_client=mlflow_client,
thresholds={
"precision": 0.85, # Don't flag more than 15% false positives
"recall": 0.70, # Catch at least 70% of actual fraud
"auc_roc": 0.92,
"inference_p99_ms": 50 # Must be fast enough for real-time scoring
}
)Every model version is evaluated against these gates before it can reach production. This prevents model quality regressions and enforces latency SLAs at the model layer, not just the infrastructure layer.
*Zak Hassan is a Staff SRE specializing in ML infrastructure reliability, data platform engineering, and AI-powered operations. Find him at zakhassan.com or on LinkedIn.*
Topic Paths