Skip to content
Aryanto
Back to Blog

Real-Time Feature Stores: Architecture, Consistency, and Production Patterns

A production-focused technical deep-dive from 9+ years of hands-on data science experience.

• Aryanto, M.Si
feature-stores mlops architecture

Real-Time Feature Stores: Architecture, Consistency, and Production Patterns

In 2021, I was building a fraud detection system that needed to score transactions in under 50 milliseconds. The model itself took 5ms to run inference. The other 45ms? Spent fetching and computing features—user transaction history, merchant risk scores, device fingerprints, behavioral patterns. The model was fast; the features were slow. That project taught me the most important lesson in production ML: the feature problem is harder than the model problem.

A feature store solves this problem by precomputing, storing, and serving features with low latency. It is the bridge between your data engineering infrastructure and your ML models—the critical layer that ensures your models have access to the right features at the right time, in both training and serving.

After nine years of building ML systems, I have used feature stores in fraud detection, recommendation systems, dynamic pricing, and risk scoring. This is a practitioner’s guide to real-time feature stores: the architecture decisions, the consistency challenges, and the patterns that actually work in production.

The Problem Feature Stores Solve

Without a feature store, the typical ML architecture looks like this:

Training time: Data scientists write ad-hoc SQL queries or Python scripts to compute features from raw data. Features are materialized into training datasets. This process is often manual, non-reproducible, and differs from how features are computed at serving time.

Serving time: Engineers write separate code (often in a different language) to compute the same features for real-time predictions. This code must be kept in sync with the training feature logic—an inherently fragile process.

The result: training-serving skew. The model sees slightly different features during training versus production, leading to degraded performance, subtle bugs, and hard-to-diagnose failures.

A feature store eliminates this by providing a single source of truth for feature computation, with APIs for both training (batch) and serving (online) access.

Architecture: Batch and Streaming

A production feature store must handle two fundamentally different compute patterns:

Batch Features

Computed from historical data on a schedule (hourly, daily, weekly). Examples:

  • User lifetime value
  • Average transaction amount over 30 days
  • Number of support tickets in the last 90 days
  • Product rating distribution

These features are computed by batch jobs (Spark, Beam, dbt) and stored in the feature store’s offline store for training and in the online store for serving.

Streaming Features

Computed from real-time event streams. Examples:

  • Number of transactions in the last 5 minutes
  • Current session page views
  • Real-time inventory levels
  • Active user count in the last 60 seconds

These features are computed by streaming processors (Flink, Kafka Streams, Spark Streaming) and written directly to the online store.

The Architecture

┌─────────────────────────────────────────────────────────┐
│                     Feature Store                        │
│                                                          │
│  ┌──────────────┐    ┌──────────────┐                   │
│  │ Batch        │    │ Streaming    │                   │
│  │ Compute      │    │ Compute      │                   │
│  │ (Spark/dbt)  │    │ (Flink/KS)   │                   │
│  └──────┬───────┘    └──────┬───────┘                   │
│         │                    │                            │
│         ▼                    ▼                            │
│  ┌──────────────┐    ┌──────────────┐                   │
│  │ Offline Store│    │ Online Store │                   │
│  │ (Parquet/    │    │ (Redis/      │                   │
│  │  BigQuery)   │    │  DynamoDB)   │                   │
│  └──────┬───────┘    └──────┬───────┘                   │
│         │                    │                            │
│         ▼                    ▼                            │
│  ┌──────────────┐    ┌──────────────┐                   │
│  │ Training API │    │ Serving API  │                   │
│  │ (batch)      │    │ (low-latency)│                   │
│  └──────────────┘    └──────────────┘                   │
└─────────────────────────────────────────────────────────┘

Feature Store Implementations

Feast (Open Source)

Feast is the most widely adopted open-source feature store. It provides a consistent API for defining, storing, and retrieving features.

from feast import FeatureStore, Entity, FeatureView, Field
from feast.types import Float32, Int64
from feast.infra.offline_stores.file_source import FileSource
from feast.infra.online_stores.redis import RedisOnlineStoreConfig
from datetime import timedelta

# Define entity
user = Entity(
    name="user",
    join_keys=["user_id"],
)

# Define batch feature source
user_stats_source = FileSource(
    path="data/user_stats.parquet",
    timestamp_field="event_timestamp",
)

# Define feature view
user_stats_fv = FeatureView(
    name="user_stats",
    entities=[user],
    ttl=timedelta(days=1),
    schema=[
        Field(name="total_transactions", dtype=Int64),
        Field(name="avg_transaction_amount", dtype=Float32),
        Field(name="days_since_last_transaction", dtype=Int64),
        Field(name="preferred_category", dtype=Int64),
    ],
    source=user_stats_source,
)

# Apply to feature store
store = FeatureStore(repo_path=".")
store.apply([user, user_stats_fv])

# Retrieve features for training (offline)
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_stats:total_transactions",
        "user_stats:avg_transaction_amount",
        "user_stats:days_since_last_transaction",
    ],
).to_df()

# Retrieve features for serving (online)
online_features = store.get_online_features(
    features=[
        "user_stats:total_transactions",
        "user_stats:avg_transaction_amount",
    ],
    entity_rows=[{"user_id": "user_123"}],
).to_dict()

Custom Feature Store with Redis

For more control, I often build a custom feature store backed by Redis:

import redis
import json
import pickle
from datetime import datetime

class FeatureStore:
    def __init__(self, redis_host="localhost", redis_port=6379):
        self.redis = redis.Redis(host=redis_host, port=redis_port, db=0)
        self.pipeline = None

    def set_features(self, entity_id: str, feature_group: str,
                     features: dict, ttl: int = 86400):
        """Store features for an entity."""
        key = f"features:{feature_group}:{entity_id}"
        features["_updated_at"] = datetime.utcnow().isoformat()
        self.redis.setex(key, ttl, json.dumps(features))

    def get_features(self, entity_id: str, feature_group: str) -> dict:
        """Retrieve features for an entity."""
        key = f"features:{feature_group}:{entity_id}"
        data = self.redis.get(key)
        if data is None:
            return {}
        return json.loads(data)

    def get_multi_features(self, entity_ids: list[str],
                           feature_groups: list[str]) -> list[dict]:
        """Retrieve features for multiple entities efficiently."""
        pipe = self.redis.pipeline()
        keys = []
        for entity_id in entity_ids:
            for group in feature_groups:
                key = f"features:{group}:{entity_id}"
                pipe.get(key)
                keys.append((entity_id, group))

        results = pipe.execute()

        features_by_entity = {}
        for (entity_id, group), result in zip(keys, results):
            if entity_id not in features_by_entity:
                features_by_entity[entity_id] = {}
            if result is not None:
                features_by_entity[entity_id][group] = json.loads(result)

        return [
            features_by_entity.get(eid, {})
            for eid in entity_ids
        ]

    def set_batch_features(self, feature_group: str,
                           features_by_entity: dict[str, dict]):
        """Batch write features for multiple entities."""
        pipe = self.redis.pipeline()
        for entity_id, features in features_by_entity.items():
            key = f"features:{feature_group}:{entity_id}"
            features["_updated_at"] = datetime.utcnow().isoformat()
            pipe.setex(key, 86400, json.dumps(features))
        pipe.execute()

Feature Computation Layer

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

class FeatureComputer:
    def __init__(self, spark: SparkSession):
        self.spark = spark

    def compute_user_features(self, transactions_df, lookback_days=30):
        """Compute user-level features from transactions."""
        cutoff = F.date_sub(F.current_date(), lookback_days)
        recent = transactions_df.filter(F.col("transaction_date") >= cutoff)

        user_features = recent.groupBy("user_id").agg(
            F.count("*").alias("transaction_count"),
            F.sum("amount").alias("total_amount"),
            F.avg("amount").alias("avg_amount"),
            F.stddev("amount").alias("stddev_amount"),
            F.max("amount").alias("max_amount"),
            F.countDistinct("merchant_id").alias("unique_merchants"),
            F.countDistinct("category").alias("unique_categories"),
            F.datediff(
                F.current_date(),
                F.max("transaction_date")
            ).alias("days_since_last_transaction"),
            # Time-based features
            F.sum(F.when(
                F.hour("transaction_time").between(22, 23) |
                F.hour("transaction_time").between(0, 5),
                1
            ).otherwise(0)).alias("nighttime_transactions"),
        )

        # Derived features
        user_features = user_features.withColumn(
            "nighttime_ratio",
            F.col("nighttime_transactions") / F.col("transaction_count")
        ).withColumn(
            "avg_merchants_per_txn",
            F.col("unique_merchants") / F.col("transaction_count")
        )

        return user_features

    def compute_streaming_features(self, event_stream):
        """Compute features from streaming events."""
        from pyspark.sql.functions import window

        # 5-minute tumbling window aggregation
        windowed_features = event_stream \
            .withWatermark("event_time", "10 minutes") \
            .groupBy(
                F.col("user_id"),
                window(F.col("event_time"), "5 minutes")
            ).agg(
                F.count("*").alias("event_count_5m"),
                F.sum("amount").alias("total_amount_5m"),
                F.countDistinct("event_type").alias("unique_event_types_5m"),
            )

        return windowed_features

Training-Serving Consistency

The most critical challenge in feature stores is ensuring that features computed during training match features computed during serving. This is harder than it sounds.

The Problem

Batch training: Features are computed over a historical window. For example, “total_transactions_last_30_days” computed on 2025-01-15 includes transactions from 2024-12-16 to 2025-01-15.

Online serving: The same feature is computed from the current state. But if you compute it at 2025-01-15 14:30:00, it might include a transaction that arrived at 14:29:59—a transaction that would not have been available at prediction time in the training data.

This subtle difference can cause significant model degradation.

Solution 1: Point-in-Time Correctness

Feature stores must support point-in-time feature retrieval—getting feature values as they were at a specific timestamp, not as they are now.

# Feast's point-in-time join
training_df = store.get_historical_features(
    entity_df=entity_df,  # Contains user_id and event_timestamp
    features=[
        "user_stats:total_transactions",
        "user_stats:avg_transaction_amount",
    ],
).to_df()

The entity_df contains both the entity ID and the timestamp for each training example. The feature store returns feature values as they were at that timestamp, preventing data leakage.

Solution 2: Feature Materialization with Timestamps

class TimestampedFeatureStore:
    def __init__(self, redis_client):
        self.redis = redis_client

    def set_features(self, entity_id: str, feature_group: str,
                     features: dict, timestamp: datetime):
        """Store features with explicit timestamp."""
        key = f"features:{feature_group}:{entity_id}"

        # Store current version
        features["_timestamp"] = timestamp.isoformat()
        self.redis.setex(key, 86400, json.dumps(features))

        # Store in time-series for point-in-time retrieval
        ts_key = f"features_ts:{feature_group}:{entity_id}"
        self.redis.zadd(ts_key, {
            json.dumps(features): timestamp.timestamp()
        })
        # Keep only last 90 days
        cutoff = (timestamp - timedelta(days=90)).timestamp()
        self.redis.zremrangebyscore(ts_key, "-inf", cutoff)

    def get_features_at_time(self, entity_id: str, feature_group: str,
                             at_time: datetime) -> dict:
        """Get features as they were at a specific time."""
        ts_key = f"features_ts:{feature_group}:{entity_id}"

        # Get the most recent entry before at_time
        results = self.redis.zrevrangebyscore(
            ts_key,
            at_time.timestamp(),
            "-inf",
            start=0,
            num=1,
        )

        if not results:
            return {}

        return json.loads(results[0])

Solution 3: Feature Validation

Always validate feature consistency between offline and online stores:

def validate_feature_consistency(
    offline_store,
    online_store,
    entity_ids: list[str],
    feature_groups: list[str],
    sample_size: int = 1000,
) -> dict:
    """Validate that offline and online features match."""
    import random

    sampled_ids = random.sample(entity_ids, min(sample_size, len(entity_ids)))

    mismatches = []
    for entity_id in sampled_ids:
        for group in feature_groups:
            offline_features = offline_store.get_features(entity_id, group)
            online_features = online_store.get_features(entity_id, group)

            for key in offline_features:
                if key.startswith("_"):
                    continue
                offline_val = offline_features[key]
                online_val = online_features.get(key)

                if online_val is None:
                    mismatches.append({
                        "entity_id": entity_id,
                        "feature": key,
                        "offline": offline_val,
                        "online": "MISSING",
                    })
                elif abs(float(offline_val) - float(online_val)) > 1e-6:
                    mismatches.append({
                        "entity_id": entity_id,
                        "feature": key,
                        "offline": offline_val,
                        "online": online_val,
                    })

    return {
        "total_checked": len(sampled_ids) * len(feature_groups),
        "mismatches": len(mismatches),
        "mismatch_rate": len(mismatches) / (len(sampled_ids) * len(feature_groups)),
        "details": mismatches[:10],  # First 10 mismatches
    }

Latency Optimization

For real-time feature serving, latency is critical. Here are the optimization techniques that matter:

1. Precompute Everything Possible

Never compute features at serving time if you can precompute them. The 30-day transaction count should be a stored value, not a query.

# BAD: Compute at serving time (slow)
def get_user_features_slow(user_id, db_connection):
    result = db_connection.execute("""
        SELECT COUNT(*), AVG(amount)
        FROM transactions
        WHERE user_id = %s
        AND transaction_date >= NOW() - INTERVAL 30 DAY
    """, user_id)
    return result

# GOOD: Precompute and store (fast)
def get_user_features_fast(user_id, feature_store):
    return feature_store.get_features(user_id, "user_stats")

2. Batch Retrieval

When scoring multiple entities, retrieve features in a single batch operation:

# BAD: One call per entity
features = [store.get_features(uid, "user_stats") for uid in user_ids]

# GOOD: Single batch call
features = store.get_multi_features(user_ids, ["user_stats", "transaction_stats"])

3. Feature Caching

For features that change slowly, add a local cache:

from functools import lru_cache
from cachetools import TTLCache

class CachedFeatureStore:
    def __init__(self, feature_store, cache_ttl=300, cache_size=10000):
        self.store = feature_store
        self.cache = TTLCache(maxsize=cache_size, ttl=cache_ttl)

    def get_features(self, entity_id: str, feature_group: str) -> dict:
        cache_key = f"{entity_id}:{feature_group}"
        if cache_key in self.cache:
            return self.cache[cache_key]

        features = self.store.get_features(entity_id, feature_group)
        self.cache[cache_key] = features
        return features

4. Data Type Optimization

Use compact data types to reduce serialization overhead:

# Use integers instead of strings for categorical features
# Use float32 instead of float64 where precision allows
# Use compact serialization formats (MessagePack instead of JSON)

import msgpack

class CompactFeatureStore:
    def set_features(self, entity_id, group, features):
        key = f"features:{group}:{entity_id}"
        packed = msgpack.packb(features, use_bin_type=True)
        self.redis.setex(key, 86400, packed)

    def get_features(self, entity_id, group):
        key = f"features:{group}:{entity_id}"
        data = self.redis.get(key)
        if data is None:
            return {}
        return msgpack.unpackb(data, raw=False)

Feature Versioning

As models evolve, feature definitions change. A production feature store must support versioning.

class VersionedFeatureStore:
    def __init__(self, store):
        self.store = store

    def set_features(self, entity_id, group, features, version="v1"):
        key = f"features:{group}:{version}:{entity_id}"
        self.store.redis.setex(key, 86400, json.dumps(features))

    def get_features(self, entity_id, group, version="v1"):
        key = f"features:{group}:{version}:{entity_id}"
        data = self.store.redis.get(key)
        return json.loads(data) if data else {}

    def get_latest_features(self, entity_id, group):
        """Get features from the latest version."""
        # Try versions in reverse order
        for version in ["v3", "v2", "v1"]:
            features = self.get_features(entity_id, group, version)
            if features:
                return features
        return {}

Feature Registry

Maintain a registry of feature definitions, versions, and metadata:

from dataclasses import dataclass
from datetime import datetime

@dataclass
class FeatureDefinition:
    name: str
    group: str
    version: str
    description: str
    data_type: str
    computation_logic: str
    created_at: datetime
    owner: str
    tags: list[str]

class FeatureRegistry:
    def __init__(self):
        self.features = {}

    def register(self, feature: FeatureDefinition):
        key = f"{feature.group}:{feature.name}:{feature.version}"
        self.features[key] = feature

    def get_definition(self, group, name, version="latest"):
        if version == "latest":
            # Find latest version
            matching = [
                f for k, f in self.features.items()
                if f.group == group and f.name == name
            ]
            return max(matching, key=lambda f: f.version) if matching else None
        return self.features.get(f"{group}:{name}:{version}")

    def list_features(self, group=None, tags=None):
        features = self.features.values()
        if group:
            features = [f for f in features if f.group == group]
        if tags:
            features = [f for f in features if any(t in f.tags for t in tags)]
        return list(features)

Architectural Patterns

Pattern 1: Lambda Architecture for Features

Batch Layer → Offline Store → Training
     ↓
Streaming Layer → Online Store → Serving
     ↓
Serving Layer → Model → Prediction

The batch layer computes features from historical data (daily/hourly). The streaming layer computes features from real-time events. The online store merges both for serving.

Pattern 2: Kappa Architecture for Features

Event Stream → Feature Processor → Online Store → Model → Prediction
                    ↓
              Offline Store → Training

All features are computed from the event stream. Batch features are computed by replaying historical events. This simplifies the architecture but requires a robust event streaming platform.

Pattern 3: Request-Enriched Features

Request → Feature Store Lookup → Feature Enrichment → Model → Prediction
     ↓
  Contextual Features (computed at request time)

Some features cannot be precomputed—they depend on the current request context (e.g., time of day, device type, current location). These are computed at request time and merged with precomputed features.

def enrich_with_contextual_features(precomputed: dict,
                                     request: dict) -> dict:
    """Add features computed at request time."""
    from datetime import datetime

    now = datetime.utcnow()
    return {
        **precomputed,
        "hour_of_day": now.hour,
        "day_of_week": now.weekday(),
        "is_weekend": int(now.weekday() >= 5),
        "device_type": request.get("device_type", "unknown"),
        "is_mobile": int(request.get("device_type") == "mobile"),
        "request_timestamp": now.isoformat(),
    }

Lessons from Production

  1. Feature engineering is the bottleneck, not model training: I spend 70% of my time on features and 30% on models. The feature store is the most impactful infrastructure investment.

  2. Start with batch, add streaming incrementally: Most features do not need real-time computation. Start with batch features and add streaming only when latency requirements demand it.

  3. Monitor feature freshness: Stale features are worse than no features. Set up monitoring to alert when features have not been updated within their expected TTL.

  4. Test features like you test code: Feature definitions should be version-controlled, tested, and deployed through CI/CD. A broken feature computation can silently degrade model performance.

  5. Document feature semantics: Every feature should have a clear definition, computation logic, and expected distribution. When a feature’s value looks wrong, you need to know what “right” looks like.

  6. Plan for backfill: When you add a new feature, you need to backfill it for historical training data. Design your feature computation pipeline to support point-in-time backfill.

Conclusion

A feature store is not optional for production ML at scale. It is the critical infrastructure that ensures your models have consistent, fresh, and correctly computed features in both training and serving.

The key architectural decisions are: batch vs. streaming computation, offline vs. online storage, and how to ensure training-serving consistency. Get these right, and your feature store will be the foundation that makes everything else in your ML system work reliably.

Start simple: a Redis-backed feature store with batch computation handles most use cases. Add streaming, caching, versioning, and advanced consistency guarantees as your system matures. The patterns described here are battle-tested across multiple production systems and will serve you well as you scale.