Data Feature Store

Feature Store

Purpose

Manage ML features through their lifecycle: define feature definitions, compute from batch/streaming sources, serve at low latency for online inference, and generate point-in-time correct training datasets.

Agent Protocol

Trigger

Exact user phrases: "feature store", "Feast", "Tecton", "feature engineering", "feature serving", "feature registry", "point-in-time join", "online features", "offline features", "feature pipeline", "feature retrieval", "feature management", "ML feature".

Input Context

Before activating, verify:

• ML framework (PyTorch, TensorFlow, scikit-learn)
• Inference mode (batch scoring, real-time API)
• Feature sources (data warehouse, streaming, real-time APIs)
• Infrastructure (Kubernetes, cloud provider, on-prem)
• Online serving requirements (latency, throughput, freshness)
• Existing feature definitions location

Output Artifact

Feature store configuration with Feast deployment, feature definitions, serving infrastructure, and training dataset generation pipeline.

Response Format

# Feast feature definitions
# Serving config

# Feature retrieval for training
# Online feature serving
# Point-in-time join

# Feature engineering queries

No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output — why use many token when few do trick.

Completion Criteria

• Feast or Tecton deployed with offline and online store
• Feature definitions registered with types, sources, and owners
• Feature engineering pipeline producing batch and streaming features
• Point-in-time correct training dataset generation working
• Online serving endpoint providing features under 10ms p99
• Feature registry browsable for discovery and documentation
• Feature validation and monitoring configured

Max Response Length

300 lines of code and configuration.

Feast Feature Definitions

Feature Repository Structure

feature_repo/
├── feature_store.yaml       # Feast config
├── features/
│   ├── user_features.py     # User-related features
│   ├── order_features.py    # Order-related features
│   └── merchant_features.py # Merchant features
└── analysis/
    └── feature_stats.py     # Feature distribution analysis

feature_store.yaml

project: ml_features
provider: gcp
registry:
  path: gs://ml-feature-registry/registry.db
  cache_ttl_seconds: 3600
online_store:
  type: redis
  connection_string: redis://redis-feast:6379
offline_store:
  type: bigquery
  dataset: feast_offline

Feature Definition

# features/user_features.py
from datetime import timedelta
from feast import (
    Entity, FeatureView, Field, FileSource, ValueType
)
from feast.types import Float32, Int32, String

user = Entity(
    name="user_id",
    value_type=ValueType.INT64,
    description="Customer identifier",
)

user_source = FileSource(
    path="gs://ml-data/features/users/*.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_at",
)

user_features = FeatureView(
    name="user_features",
    entities=[user],
    ttl=timedelta(days=30),
    schema=[
        Field(name="user_id", dtype=Int32),
        Field(name="total_orders_30d", dtype=Int32),
        Field(name="avg_order_value_30d", dtype=Float32),
        Field(name="days_since_last_order", dtype=Int32),
        Field(name="customer_tenure_days", dtype=Int32),
        Field(name="preferred_category", dtype=String),
    ],
    source=user_source,
    tags={"team": "data-science", "domain": "user"},
)

Point-in-Time Joins

Training Dataset Generation

import pandas as pd
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")
entity_df = pd.DataFrame.from_dict({
    "user_id": [1001, 1002, 1003, 1004],
    "event_timestamp": [
        "2026-05-01 12:00:00",
        "2026-05-01 13:00:00",
        "2026-05-02 09:00:00",
        "2026-05-03 15:00:00",
    ],
})

training_data = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_features:total_orders_30d",
        "user_features:avg_order_value_30d",
        "user_features:days_since_last_order",
        "user_features:customer_tenure_days",
        "order_features:order_count_7d",
        "order_features:total_spend_7d",
    ],
).to_df()

# Feast ensures each feature value is the value as-of the event_timestamp
# No data leaks — features from the future are never included
print(training_data.head())

Point-in-Time Join Mechanics

Entity row: user_id=1001, event_timestamp=2026-05-01 12:00
    ↓
Feast looks up user_features table:
    Finds most recent row with event_timestamp <= 2026-05-01 12:00
    Returns total_orders_30d=12 (computed at 2026-05-01 06:00)
    Does NOT use total_orders_30d=14 (computed at 2026-05-01 14:00 — future)
    ↓
Result: clean training example with no future data leakage

Online Feature Serving

Low-Latency Retrieval

from feast import FeatureStore
import time

store = FeatureStore(repo_path="./feature_repo")

def get_online_features(user_ids: list[int]) -> dict:
    """Retrieve latest features for real-time inference."""
    start = time.time()

    features = store.get_online_features(
        features=[
            "user_features:total_orders_30d",
            "user_features:avg_order_value_30d",
            "user_features:days_since_last_order",
        ],
        entity_rows=[{"user_id": uid} for uid in user_ids],
    ).to_dict()

    latency = (time.time() - start) * 1000
    return {"features": features, "latency_ms": latency}

# Usage in prediction API
@app.post("/predict")
def predict(request: PredictionRequest):
    features = get_online_features(request.user_ids)
    predictions = model.predict(features["features"])
    return {"predictions": predictions.tolist()}

Redis Online Store Config

# Redis cluster for production online serving
online_store:
  type: redis
  connection_string: redis://redis-cluster:6379
  key_ttl_seconds: 86400  # 24 hour TTL on feature keys
  password_encrypted: ${FEAST_REDIS_PASSWORD}

# Optional: Redis Sentinel
  redis_type: redis_cluster
  sentinel_master: feast-master
  sentinel_set:
    - host: redis-sentinel-0:26379
    - host: redis-sentinel-1:26379

Feast Deployment

Docker Compose

version: '3.8'
services:
  feast-server:
    image: feastdev/feature-server:0.38
    ports:
      - "6566:6566"
    environment:
      FEAST_FEATURE_STORE: /etc/feast/feature_store.yaml
      FEAST_REDIS_HOST: redis
      FEAST_REDIS_PORT: 6379
    volumes:
      - ./feature_repo:/etc/feast

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: redis-server --appendonly yes

  feast-ui:
    image: feastdev/feast-ui:0.38
    ports:
      - "8888:80"
    environment:
      FEAST_UI_REGISTRY_URL: gs://ml-feature-registry/registry.db

Feature Engineering Pipelines

Batch Feature Pipeline (Spark)

# spark_feature_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, sum, avg, datediff, current_date, max as spark_max

spark = SparkSession.builder.appName("user-features").getOrCreate()

orders = spark.table("warehouse.orders")
customers = spark.table("warehouse.customers")

features = orders.join(customers, "customer_id").groupBy("customer_id").agg(
    count(when(col("order_date") >= date_sub(current_date(), 30), True)).alias("total_orders_30d"),
    avg(when(col("order_date") >= date_sub(current_date(), 30), col("amount"))).alias("avg_order_value_30d"),
    datediff(current_date(), spark_max("order_date")).alias("days_since_last_order"),
)

features.write.mode("overwrite").parquet("gs://ml-data/features/users/")

Streaming Feature Pipeline (Kafka + Flink)

# streaming_user_features.py
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes

env = StreamExecutionEnvironment.get_execution_context()
t_env = StreamTableEnvironment.create(env)

t_env.execute_sql("""
    CREATE TABLE user_events (
        user_id INT,
        event_type STRING,
        amount DOUBLE,
        event_time TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user-events',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'json'
    )
""")

t_env.execute_sql("""
    CREATE TABLE redis_sink (
        user_id INT,
        total_orders_5m BIGINT,
        total_amount_5m DOUBLE,
        PRIMARY KEY (user_id) NOT ENFORCED
    ) WITH (
        'connector' = 'redis',
        'format' = 'json',
        'redis.host' = 'redis-feast',
        'redis.port' = '6379'
    )
""")

t_env.execute_sql("""
    INSERT INTO redis_sink
    SELECT
        user_id,
        COUNT(*) as total_orders_5m,
        SUM(amount) as total_amount_5m
    FROM user_events
    WHERE event_type = 'order_placed'
    GROUP BY user_id, TUMBLE(event_time, INTERVAL '5' MINUTE)
""")

Feature Validation and Monitoring

Feature Validation

# feature_stats.py
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")
fv = store.get_feature_view("user_features")
features = store.get_historical_features(
    entity_df=entity_df,
    features=[f"user_features:{f.name}" for f in fv.schema],
).to_df()

checks = {
    "total_orders_30d": {
        "min": 0,
        "max": 1000,
        "null_pct": 0,
    },
    "avg_order_value_30d": {
        "min": 0.0,
        "max": 50000.0,
        "null_pct": 0.05,
    },
    "days_since_last_order": {
        "min": 0,
        "max": 365,
        "null_pct": 0.1,
    },
}

for col, rules in checks.items():
    null_pct = features[col].isnull().mean()
    assert null_pct <= rules["null_pct"], f"{col}: null {null_pct:.2%} > {rules['null_pct']:.0%}"
    assert features[col].min() >= rules["min"], f"{col}: min {features[col].min()} < {rules['min']}"
    assert features[col].max() <= rules["max"], f"{col}: max {features[col].max()} > {rules['max']}"

Feast Architecture

feast_architecture:
  feature_repository:
    path: "features/"
    components:
      - feature_store.yaml  # Infrastructure config
      - features/           # Feature definitions (Python)
      - requirements.txt    # Feast version + dependencies
      - .feastignore        # Files to exclude from registry
  
  feature_store_yaml:
    project: "customer_features"
    registry: "gs://feature-registry/registry.db"  # Shared registry
    provider: "gcp"
    
    offline_store:
      type: "bigquery"
      dataset: "feature_store"
    
    online_store:
      type: "redis"
      connection_string: "redis://redis-cluster:6379"

  serving_patterns:
    batch_serving:
      - "Feast.get_historical_features(entity_df, features) → Pandas/Spark DF"
      - "Used for: training data generation, batch scoring"
    
    online_serving:
      - "feast_client.get_online_features(features, entity_keys) → Proto/JSON"
      - "Used for: real-time model inference, API serving"
      - "Latency: p99 < 10ms for 100 entity keys"
    
    streaming_serving:
      - "Push features to online store via Feast push API"
      - "Streaming feature computation → push → online store"
      - "Used for: real-time features in streaming ML"

Feature Serving Decision Tree

Feature freshness requirement?
├── Historical data only (batch training)
│   └── Offline store: BigQuery/Redshift/Snowflake (point-in-time joins)
├── Batch serving (daily/hourly predictions)
│   ├── Materialized offline → batch inference
│   └── Online store fallback for recent features
├── Real-time serving (sub-second)
│   ├── Pre-computed features → Online store: Redis/DynamoDB
│   ├── Stream-computed features → Feature push API → Online store
│   └── Feature TTL: 24h-7d for freshness-staleness tradeoff
└── Streaming (event-time features)
    ├── Streaming feature computation (Kafka/Flink)
    ├── Push to online store via streaming sink
    └── Feature TTL: minutes-hours for event-time features

Rules

• All features have point-in-time correctness — no future data leakage
• Entity definitions use the same ID type across all feature views
• Online store supports p99 < 10ms for batch of 100 entity keys
• Offline store uses columnar format (Parquet) for training data generation
• Feature TTL prevents staleness — stale features not served online
• Feature registry is the single source of truth for all feature definitions
• Feature validation runs before materialization
• Monitoring tracks feature freshness, serving latency, and retrieval errors
• Feature engineering pipelines are idempotent and incremental
• Match serving store to latency requirements — batch for training, online for inference
• Use Feast feature repository for version-controlled feature definitions

References

• references/feast-setup-guide.md — Feast Setup Guide
• references/feature-engineering-pipeline.md — Feature Engineering Pipeline Reference
• references/feature-registry.md — Feature Registry & Governance
• references/feature-validation-monitoring.md — Feature Validation and Monitoring
• references/offline-feature-computation.md — Offline Feature Computation
• references/online-serving.md — Online Feature Serving Reference

Architecture Decision Trees

Feature Store Selection
├── ML framework ecosystem?
│   ├── Python-heavy → Feast (Python-native, great with scikit-learn/XGBoost)
│   ├── Spark ecosystem → Feature Store on Databricks / SageMaker
│   └── Multi-language → Tecton (SaaS, Python + SQL + Spark)
├── Online serving required?
│   ├── Yes → Feast + Redis/DynamoDB (low-latency serving)
│   └── No → Offline-only (Feast BigQuery/Redshift)
├── Point-in-time correct joins?
│   ├── Yes → Feast (temporal join built-in)
│   └── No → Custom SQL with window functions
└── Feature sharing across teams?
    ├── Yes → Centralized registry (Feast with GCS/S3 registry)
    └── No → Per-team feature definitions

Decision criteria: Evaluate online/offline serving needs, team ML maturity, feature reuse goals, and infrastructure compatibility.

Implementation Patterns

Feast Feature Definition

# feature_store/features.py
from feast import Entity, FeatureView, ValueType, FeatureService
from feast.infra.offline_stores.bigquery_source import BigQuerySource

customer = Entity(
    name="customer_id",
    value_type=ValueType.INT64,
    description="Customer identifier",
)

order_features = FeatureView(
    name="order_features",
    entities=[customer],
    ttl="90d",
   
    source=BigQuerySource(
        query="SELECT customer_id, order_count, avg_order_value, last_order_date FROM feature_store.order_aggregates"
    ),
    tags={"team": "fraud", "tier": "critical"},
)

Online Feature Serving

# feature_store/online_serving.py
from feast import FeatureStore

class FeatureServingClient:
    def __init__(self, repo_path: str = "./feature_repo"):
        self.store = FeatureStore(repo_path)

    def get_online_features(self, features: list[str], entities: list[dict]) -> dict:
        feature_vector = self.store.get_online_features(
            features=features,
            entities=entities,
        ).to_dict()
        return feature_vector

    def get_historical_features(self, entity_df, features: list[str]):
        job = self.store.get_historical_features(
            entity_df=entity_df,
            features=features,
        )
        return job.to_df()

Production Considerations

• Online store sizing: Provision Redis cluster with enough memory for all active features (~2x expected size for overhead).
• Feature freshness: Set TTL per feature view based on update frequency; stale features auto-expire.
• Feature validation: Validate feature distributions against training baseline; alert on drift > 2σ.
• Registry synchronization: Sync feature registry across environments (dev → staging → prod) via CI/CD.
• Point-in-time correctness: Ensure training datasets use point-in-time joins to prevent data leakage.
• Monitoring: Track online serving latency (p99 < 10ms), feature retrieval rate, and cache hit rate.

Anti-Patterns

Performance Optimization

• Online store caching: Add local cache (Redis on same node) for frequently accessed features; TTL 60s.
• Batch feature materialization: Pre-compute batch features hourly; use incremental materialization.
• Vectorized serving: Batch entity requests into groups of 100 for reduced network round-trips.
• Feature embedding: Pre-compute embedding features offline; serve from vector DB (FAISS, Pinecone).
• Parquet optimization: Store offline feature data in Parquet with partition pruning by date and entity key.

Security Considerations

• Feature ACLs: Restrict feature access by team/service account; sensitive features require approval.
• Online store encryption: Enable Redis encryption in transit (TLS) and at rest (AES-256).
• Registry protection: Sign feature registry commits; verify signatures in CI/CD deployment.
• Audit trail: Log all feature registry changes, materialization runs, and online serving requests.
• Compliance: Tag features containing PII; strip or hash PII features in non-production environments.

Handoff

streaming for real-time feature computation with Kafka and Flink etl-pipeline for batch feature engineering orchestration with Airflow