You’ve built a robust machine learning model, meticulously engineered its features, and perhaps even deployed it with a feature store to ensure consistency between training and serving. Everything seems stable. Yet, lurking beneath the surface is a silent killer: input data drift. This isn't about your model making bad predictions, but about the very nature of the data it receives changing over time, subtly degrading performance without a single error message. For ML engineers and data scientists, understanding and proactively detecting this drift is paramount to preventing catastrophic model failures. In this post, I'll walk you through building a practical monitoring system to detect input data drift in text features, using real-world data from the GitHub Engineering blog, demonstrating how to catch these shifts before they impact your users.
Key Takeaways
- Input data drift can silently degrade ML model performance, even with robust feature engineering and serving pipelines.
- The Kolmogorov-Smirnov (KS) test is a pragmatic statistical tool for comparing feature distributions between baseline and live data streams.
- Effective drift detection requires defining interpretable numerical features from raw data, like document length or keyword frequencies, that capture underlying data characteristics.
- Establishing a clear baseline from historical, known-good data is crucial for meaningful comparisons with incoming live data.
- A simple alerting mechanism based on statistical significance (e.g., p-value thresholds) can proactively flag potential drift for investigation.
The Problem
Imagine your fraud detection model, trained on a specific pattern of financial transactions. If the underlying behavior of fraudsters changes—new methods emerge, or typical transaction amounts shift—your model will start to perform poorly. Even if your feature store ensures the features are calculated identically between training and inference, the *distribution* of those features can silently diverge from what the model was trained on. This is data drift, and it's a critical, often overlooked, aspect of MLOps. Without a mechanism to detect these shifts, you're flying blind, waiting for a drop in business metrics to tell you something is wrong, by which time, significant impact may have already occurred.
Data and Sources
To simulate a real-world, continuously updating text data stream, I'm using the GitHub Engineering RSS Feed. This feed provides a stream of technical articles, which serves as an excellent proxy for blog posts, news articles, or documentation that your ML models might process. The content and style of these articles can change over time, providing a realistic scenario for drift detection.
- GitHub Engineering RSS Feed: https://github.blog/engineering/feed/
Data accessed on 2024-07-28.
Step 1 — Establishing a Production-Like Data Stream
The first challenge is to reliably fetch and parse our data source, making it feel like a continuous stream of new information. For RSS feeds, the feedparser library is incredibly handy. It abstracts away the complexities of XML parsing and HTTP requests, giving us a structured view of the feed's entries.
Here’s how I fetch the data and extract the relevant text fields (title and summary) that our hypothetical ML model would consume:
import feedparser
import requests
import sys
def fetch_feed_data(url):
"""Fetches and parses an RSS feed, returning a list of dictionaries."""
try:
# Use requests to handle potential network issues more gracefully
response = requests.get(url, timeout=10)
response.raise_for_status() # Raise an exception for HTTP errors
feed = feedparser.parse(response.text)
if feed.bozo:
print(f"Warning: Feed parsing issues detected for {url}: {feed.bozo_exception}", file=sys.stderr)
parsed_entries = []
for entry in feed.entries:
parsed_entries.append({
'title': entry.get('title', ''),
'summary': entry.get('summary', '')
})
return parsed_entries
except requests.exceptions.RequestException as e:
print(f"Error fetching RSS feed from {url}: {e}", file=sys.stderr)
return []
except Exception as e:
print(f"An unexpected error occurred during feed parsing: {e}", file=sys.stderr)
return []
I added requests and basic error handling to make this more production-ready. Network failures are common, and gracefully handling them is crucial. The `feedparser.parse` function can take a URL directly, but using requests first gives us more control over timeouts and error responses, and then passing response.text to feedparser ensures we're dealing with the content correctly.
Step 2 — Engineering Baseline Features from Historical Data
Raw text isn't directly comparable. We need to extract numerical features that represent its core characteristics. For text data, features like document length, average word length, and the frequency of certain keywords can be powerful indicators of stylistic or thematic shifts. I'll define a function to extract these features from each article. We then apply this to a segment of our fetched data to establish a "baseline"—what our data looked like when our model was presumably performing well.
import re
from collections import Counter
def extract_text_features(text_content):
"""Extracts numerical features from a given text."""
# Combine title and summary for feature extraction
clean_text = re.sub(r'<.*?>', '', text_content).lower() # Remove HTML tags and lowercase
words = re.findall(r'\b\w+\b', clean_text) # Extract words
if not words:
return {
'word_count': 0,
'avg_word_length': 0.0,
'keyword_ai_freq': 0,
'keyword_ml_freq': 0,
'keyword_security_freq': 0,
'keyword_performance_freq': 0
}
word_count = len(words)
total_char_count = sum(len(word) for word in words)
avg_word_length = total_char_count / word_count
word_counts = Counter(words)
keyword_ai_freq = word_counts['ai']
keyword_ml_freq = word_counts['ml']
keyword_security_freq = word_counts['security']
keyword_performance_freq = word_counts['performance']
return {
'word_count': word_count,
'avg_word_length': avg_word_length,
'keyword_ai_freq': keyword_ai_freq,
'keyword_ml_freq': keyword_ml_freq,
'keyword_security_freq': keyword_security_freq,
'keyword_performance_freq': keyword_performance_freq
}
def process_entries_to_features(entries):
"""Processes a list of feed entries into a list of feature dictionaries."""
all_features = []
