Comprehensive Guide to Real-Time Feature Engineering

# Example: Using Feast for feature definition and registry
# Install Feast: pip install feast

from datetime import timedelta
from feast import Entity, FeatureView, Field, FeatureStore, RepoConfig
from feast.types import Float32, Int64

# Define an entity for our features
driver = Entity(
    name="driver_id",
    value_type=Int64,
    description="Driver ID",
)

# Define a feature view using a Parquet file as source
driver_stats_view = FeatureView(
    name="driver_stats",
    entities=[driver],
    ttl=timedelta(days=1),
    schema=[
        Field(name="avg_daily_trips", dtype=Float32),
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_hours", dtype=Float32),
    ],
    online=True,
    source=ParquetSource(path="data/driver_stats.parquet"),
    tags={"team": "driver_performance"},
)

# Initialize the feature store
repo_config = RepoConfig(
    registry="data/registry.db",
    project="driver_performance",
    provider="local",
    online_store={
        "type": "redis",
        "connection_string": "redis://localhost:6379,db=0"
    },
    offline_store={
        "type": "file"
    }
)

store = FeatureStore(config=repo_config)

# Register the entity and feature view
store.apply([driver, driver_stats_view])

# Materialize features to the online store
store.materialize_incremental(end_date=datetime.utcnow())

# Example: Using Apache Flink for real-time feature computation
# This would typically be in Java/Scala, but Python API is shown for consistency

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings
from pyflink.table.descriptors import Schema, Kafka, Json
from pyflink.table.udf import udf
from pyflink.table.expressions import col

# Set up the streaming execution environment
env = StreamExecutionEnvironment.get_execution_environment()
env_settings = EnvironmentSettings.Builder().build()
t_env = StreamTableEnvironment.create(env, environment_settings=env_settings)

# Configure Kafka source
t_env.connect(
    Kafka()
    .version("universal")
    .topic("user_events")
    .start_from_latest()
    .property("bootstrap.servers", "kafka:9092")
).with_format(
    Json()
    .fail_on_missing_field(False)
    .schema(Schema()
        .field("user_id", "BIGINT")
        .field("event_type", "STRING")
        .field("timestamp", "TIMESTAMP")
        .field("value", "DOUBLE"))
).with_schema(
    Schema()
    .field("user_id", "BIGINT")
    .field("event_type", "STRING")
    .field("event_timestamp", "TIMESTAMP")
    .field("value", "DOUBLE")
).create_temporary_table("user_events")

# Define a window aggregation for feature computation
windowed_features = t_env.sql("""
    SELECT 
        user_id,
        COUNT(*) AS event_count_1h,
        SUM(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) AS purchase_count_1h,
        AVG(value) AS avg_value_1h,
        TUMBLE_END(event_timestamp, INTERVAL '1' HOUR) AS feature_timestamp
    FROM user_events
    GROUP BY 
        user_id,
        TUMBLE(event_timestamp, INTERVAL '1' HOUR)
""")

# Register the result table for output to a feature store
t_env.connect(
    Kafka()
    .version("universal")
    .topic("user_features")
    .property("bootstrap.servers", "kafka:9092")
).with_format(
    Json()
).with_schema(
    Schema()
    .field("user_id", "BIGINT")
    .field("event_count_1h", "BIGINT")
    .field("purchase_count_1h", "BIGINT")
    .field("avg_value_1h", "DOUBLE")
    .field("feature_timestamp", "TIMESTAMP")
).create_temporary_table("user_features")

# Insert the computed features into the output table
windowed_features.insert_into("user_features")

# Execute the job
env.execute("Realtime Feature Computation")

# Example: Using the same feature transformation logic for both offline and online

import numpy as np
import pandas as pd
from typing import Dict, List

# Shared transformation logic in a common module
class FeatureTransformer:
    """Feature transformation logic shared between offline and online pipelines"""
    
    def compute_rolling_average(self, values: List[float], window: int) -> float:
        """Compute a rolling average over the specified window"""
        if not values:
            return 0.0
        window_size = min(len(values), window)
        return sum(values[-window_size:]) / window_size
    
    def compute_frequency_encoding(self, value: str, frequency_map: Dict[str, float]) -> float:
        """Replace categorical values with their frequency"""
        return frequency_map.get(value, 0.0)
    
    def normalize_value(self, value: float, min_val: float, max_val: float) -> float:
        """Min-max normalization"""
        if max_val == min_val:
            return 0.0
        return (value - min_val) / (max_val - min_val)

# Offline batch processing using the shared logic
def process_batch_data(data: pd.DataFrame, transformer: FeatureTransformer) -> pd.DataFrame:
    """Process a batch of historical data for model training"""
    result = data.copy()
    
    # Group by user and compute rolling averages
    for user_id, group in data.groupby('user_id'):
        # Sort by timestamp to ensure correct order
        group = group.sort_values('timestamp')
        purchase_amounts = group['purchase_amount'].tolist()
        
        # Compute 7-day and 30-day rolling averages
        result.loc[group.index, 'avg_purchase_7d'] = transformer.compute_rolling_average(purchase_amounts, 7)
        result.loc[group.index, 'avg_purchase_30d'] = transformer.compute_rolling_average(purchase_amounts, 30)
    
    # Other transformations...
    
    return result

# Online processing function using the same transformer
def process_online_features(user_id: str, recent_events: List[Dict], transformer: FeatureTransformer) -> Dict:
    """Process recent events for a user to generate real-time features"""
    # Sort events by timestamp
    sorted_events = sorted(recent_events, key=lambda e: e['timestamp'])
    
    # Extract purchase amounts
    purchase_amounts = [e.get('purchase_amount', 0) for e in sorted_events if 'purchase_amount' in e]
    
    # Compute the same features as in batch processing
    features = {
        'avg_purchase_7d': transformer.compute_rolling_average(purchase_amounts, 7),
        'avg_purchase_30d': transformer.compute_rolling_average(purchase_amounts, 30),
        # Other features...
    }
    
    return features

# Usage example
transformer = FeatureTransformer()

# Offline - typically in a Spark/batch job
offline_data = pd.read_parquet('historical_data.parquet')
processed_offline = process_batch_data(offline_data, transformer)
processed_offline.to_parquet('offline_features.parquet')

# Online - typically in a serving application
user_events = fetch_recent_events('user123')  # Function to get recent events from cache/store
online_features = process_online_features('user123', user_events, transformer)
# These features can now be used for real-time prediction

# Example: Building a low-latency feature serving API with FastAPI and Redis

import json
import redis
import time
from typing import Dict, List, Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# Initialize FastAPI app
app = FastAPI(title="Feature Serving API")

# Redis connection
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Feature request model
class FeatureRequest(BaseModel):
    entity_id: str
    feature_list: List[str]
    entity_type: str = "user"  # Default entity type

# Feature response model
class FeatureResponse(BaseModel):
    entity_id: str
    features: Dict[str, float]
    latency_ms: float

@app.post("/get_features", response_model=FeatureResponse)
async def get_features(request: FeatureRequest):
    """Get feature values for a specific entity"""
    start_time = time.time()
    
    # Construct Redis key format based on entity type and ID
    key_prefix = f"{request.entity_type}:{request.entity_id}"
    
    # Get all requested features in a single Redis call using pipelining
    pipe = redis_client.pipeline()
    for feature_name in request.feature_list:
        pipe.hget(key_prefix, feature_name)
    
    # Execute pipeline and get results
    feature_values = pipe.execute()
    
    # Process results
    features = {}
    for i, feature_name in enumerate(request.feature_list):
        value = feature_values[i]
        if value is None:
            # Feature not found, use default value or fallback strategy
            features[feature_name] = 0.0
        else:
            # Convert from Redis binary string and handle different data types
            try:
                features[feature_name] = float(value)
            except (ValueError, TypeError):
                # For non-numeric features, may need special handling
                features[feature_name] = 0.0
    
    # Calculate request latency
    latency_ms = (time.time() - start_time) * 1000
    
    return FeatureResponse(
        entity_id=request.entity_id,
        features=features,
        latency_ms=latency_ms
    )

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    try:
        # Verify Redis connection
        redis_client.ping()
        return {"status": "healthy", "dependencies": {"redis": "connected"}}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Service unhealthy: {str(e)}")

# Run with: uvicorn feature_serving:app --host 0.0.0.0 --port 8000 --workers 4

# Example: Feature monitoring with statistical tests and alerting

import numpy as np
import pandas as pd
from scipy import stats
from typing import Dict, List, Tuple, Optional
import time
import json
import logging
from prometheus_client import Counter, Histogram, Gauge, start_http_server

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("feature_monitoring")

# Prometheus metrics
FEATURE_REQUEST_COUNT = Counter('feature_requests_total', 'Total number of feature requests', ['feature_name'])
FEATURE_LATENCY = Histogram('feature_retrieval_latency_seconds', 'Feature retrieval latency')
FEATURE_DRIFT_GAUGE = Gauge('feature_drift_score', 'Detected drift score for features', ['feature_name'])
FEATURE_AVAILABILITY = Gauge('feature_availability_ratio', 'Ratio of successful feature retrievals', ['feature_name'])

# Start Prometheus metrics server
start_http_server(8000)

class FeatureMonitor:
    """Feature monitoring for production ML systems"""
    
    def __init__(self, reference_stats_path: str, drift_threshold: float = 0.05):
        """
        Initialize the feature monitor
        
        Args:
            reference_stats_path: Path to reference statistics from training data
            drift_threshold: p-value threshold for detecting drift
        """
        self.reference_stats = self._load_reference_stats(reference_stats_path)
        self.drift_threshold = drift_threshold
        self.recent_values = {feature: [] for feature in self.reference_stats.keys()}
        self.window_size = 1000  # Number of values to keep for each feature
        
    def _load_reference_stats(self, path: str) -> Dict[str, Dict]:
        """Load reference statistics for features"""
        with open(path, 'r') as f:
            return json.load(f)
    
    def record_feature_values(self, feature_values: Dict[str, float]):
        """Record new feature values for monitoring"""
        for feature, value in feature_values.items():
            if feature in self.recent_values:
                # Keep a sliding window of recent values
                if len(self.recent_values[feature]) >= self.window_size:
                    self.recent_values[feature].pop(0)
                self.recent_values[feature].append(value)
                
                # Update request counter metric
                FEATURE_REQUEST_COUNT.labels(feature_name=feature).inc()
    
    def check_drift(self, feature_name: str) -> Tuple[bool, float]:
        """
        Check if a feature has drifted from reference distribution
        
        Returns:
            (is_drift_detected, drift_score)
        """
        if feature_name not in self.reference_stats:
            logger.warning(f"No reference stats for feature: {feature_name}")
            return False, 0.0
            
        if len(self.recent_values[feature_name]) < 100:
            logger.info(f"Not enough values to check drift for {feature_name}")
            return False, 0.0
            
        # Get reference statistics
        ref_stats = self.reference_stats[feature_name]
        
        # Get current values
        current_values = np.array(self.recent_values[feature_name])
        
        # Perform statistical test based on distribution type
        if ref_stats.get("distribution_type") == "categorical":
            # For categorical features, use chi-square test
            ref_hist = np.array(ref_stats["histogram"]["counts"])
            ref_bins = ref_stats["histogram"]["bins"]
            
            # Count values in current data
            current_hist = np.zeros_like(ref_hist)
            for value in current_values:
                bin_idx = np.digitize(value, ref_bins) - 1
                if 0 <= bin_idx < len(current_hist):
                    current_hist[bin_idx] += 1
                    
            # Chi-square test
            _, p_value = stats.chisquare(current_hist, ref_hist)
            
        else:
            # For numeric features, use Kolmogorov-Smirnov test
            # This compares the cumulative distribution functions
            _, p_value = stats.ks_2samp(
                current_values, 
                np.random.normal(ref_stats["mean"], ref_stats["std"], size=1000)
            )
        
        # Update drift metric
        drift_score = 1.0 - p_value
        FEATURE_DRIFT_GAUGE.labels(feature_name=feature_name).set(drift_score)
        
        # Check if drift detected based on threshold
        is_drift = p_value < self.drift_threshold
        if is_drift:
            logger.warning(f"Drift detected for feature {feature_name}: p-value = {p_value:.4f}")
            
        return is_drift, drift_score
    
    def check_all_features(self) -> Dict[str, Tuple[bool, float]]:
        """Check drift for all features"""
        results = {}
        for feature in self.reference_stats.keys():
            results[feature] = self.check_drift(feature)
        return results
    
    def record_feature_latency(self, latency_ms: float):
        """Record feature retrieval latency"""
        FEATURE_LATENCY.observe(latency_ms / 1000.0)  # Convert to seconds
    
    def record_feature_availability(self, feature_name: str, is_available: bool):
        """Record feature availability"""
        availability = 1.0 if is_available else 0.0
        FEATURE_AVAILABILITY.labels(feature_name=feature_name).set(availability)

# Usage:
# monitor = FeatureMonitor("reference_stats.json")
# 
# # During feature serving:
# start_time = time.time()
# feature_values = get_features_from_store(entity_id, feature_list)
# latency_ms = (time.time() - start_time) * 1000
# 
# # Record metrics
# monitor.record_feature_values(feature_values)
# monitor.record_feature_latency(latency_ms)
# 
# # Periodically check for drift
# drift_results = monitor.check_all_features()