1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

from dataclasses import dataclass
from typing import List, Optional, Callable
import pandas as pd
import numpy as np
 
@dataclass
class QualityCheck:
    name: str
    check_fn: Callable[[pd.Series], bool]
    severity: str  # 'error', 'warning', 'info'
    description: str
 
class FeatureQualityValidator:
    def __init__(self):
        self.checks: dict[str, List[QualityCheck]] = {}
    
    def add_check(self, feature: str, check: QualityCheck):
        if feature not in self.checks:
            self.checks[feature] = []
        self.checks[feature].append(check)
    
    def validate(self, df: pd.DataFrame, feature: str) -> List[dict]:
        """Run all checks for a feature and return results"""
        if feature not in self.checks:
            return []
        
        results = []
        series = df[feature]
        
        for check in self.checks[feature]:
            try:
                passed = check.check_fn(series)
                results.append({
                    'check': check.name,
                    'passed': passed,
                    'severity': check.severity,
                    'description': check.description,
                })
            except Exception as e:
                results.append({
                    'check': check.name,
                    'passed': False,
                    'severity': 'error',
                    'description': f"Check failed with error: {str(e)}",
                })
        
        return results
 
# Define quality checks for user features
validator = FeatureQualityValidator()
 
# Completeness checks
validator.add_check("total_purchases_30d", QualityCheck(
    name="null_rate_below_threshold",
    check_fn=lambda s: s.isna().mean() < 0.05,  # <5% nulls
    severity="error",
    description="Null rate should be below 5%"
))
 
# Range checks
validator.add_check("total_purchases_30d", QualityCheck(
    name="non_negative",
    check_fn=lambda s: (s.dropna() >= 0).all(),
    severity="error",
    description="Purchase count should never be negative"
))
 
validator.add_check("total_purchases_30d", QualityCheck(
    name="reasonable_maximum",
    check_fn=lambda s: (s.dropna() <= 10000).all(),
    severity="warning",
    description="Purchase count above 10,000 is suspicious"
))
 
# Distribution checks
validator.add_check("avg_purchase_amount_30d", QualityCheck(
    name="mean_in_expected_range",
    check_fn=lambda s: 10 <= s.dropna().mean() <= 500,
    severity="warning",
    description="Average purchase amount should be $10-$500"
))
 
# Run validation
def run_feature_validation(store, feature_view: str) -> dict:
    """Run all quality checks for a feature view"""
    
    # Get recent feature data
    fv = store.get_feature_view(feature_view)
    df = get_recent_feature_data(feature_view)  # Implementation depends on store
    
    all_results = {}
    for feature in [f.name for f in fv.schema]:
        results = validator.validate(df, feature)
        all_results[feature] = results
        
        # Alert on failures
        errors = [r for r in results if not r['passed'] and r['severity'] == 'error']
        if errors:
            alert_data_quality_error(feature_view, feature, errors)
    
    return all_results

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

from feast import FeatureView, Field
from feast.types import Float64, Int64, String
from datetime import timedelta
 
# Original schema (v1)
user_features_v1 = FeatureView(
    name="user_features_v1",
    entities=[user],
    schema=[
        Field(name="purchase_count", dtype=Int64),
        Field(name="total_amount", dtype=Float64),
    ],
    source=user_source,
    online=True,
)
 
# Safe evolution: Add new fields (non-breaking)
user_features_v1_extended = FeatureView(
    name="user_features_v1",  # Same name - this is additive
    entities=[user],
    schema=[
        Field(name="purchase_count", dtype=Int64),
        Field(name="total_amount", dtype=Float64),
        # NEW FIELDS - safe to add
        Field(name="avg_amount", dtype=Float64),
        Field(name="last_purchase_date", dtype=String),
    ],
    source=user_source,
    online=True,
)
 
# Breaking change: Create new version
user_features_v2 = FeatureView(
    name="user_features_v2",  # NEW NAME for breaking changes
    description="""
    BREAKING CHANGES FROM V1:
    - purchase_count renamed to order_count (semantic clarity)
    - total_amount now includes tax (was excluding)
    - added new fields from v1 extension
    
    MIGRATION: See migration/v1_to_v2.md
    DEPRECATION: v1 will be removed on 2024-06-01
    """,
    entities=[user],
    schema=[
        Field(name="order_count", dtype=Int64),  # Renamed
        Field(name="total_amount_with_tax", dtype=Float64),  # Semantic change
        Field(name="avg_amount", dtype=Float64),
        Field(name="last_purchase_date", dtype=String),
    ],
    source=user_source_v2,
    online=True,
    tags={"version": "2.0", "deprecates": "user_features_v1"},
)
 
# Schema validation in CI/CD
def validate_schema_compatibility(old_fv, new_fv) -> dict:
    """Check if schema changes are backward compatible"""
    old_fields = {f.name: f.dtype for f in old_fv.schema}
    new_fields = {f.name: f.dtype for f in new_fv.schema}
    
    issues = []
    
    # Check for removed fields (breaking)
    removed = set(old_fields.keys()) - set(new_fields.keys())
    if removed:
        issues.append(f"BREAKING: Removed fields: {removed}")
    
    # Check for type changes (potentially breaking)
    for name in old_fields:
        if name in new_fields and old_fields[name] != new_fields[name]:
            issues.append(f"BREAKING: Type changed for {name}: {old_fields[name]} → {new_fields[name]}")
    
    # Check for additions (safe)
    added = set(new_fields.keys()) - set(old_fields.keys())
    
    return {
        'compatible': len(issues) == 0,
        'issues': issues,
        'added_fields': list(added),
        'removed_fields': list(removed),
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

import numpy as np
from scipy import stats
from typing import Tuple
import pandas as pd
 
class FeatureDriftDetector:
    """
    Detect statistical drift in feature distributions.
    Compares current distribution to a reference (training) distribution.
    """
    
    def __init__(self, reference_data: pd.DataFrame):
        self.reference = reference_data
        self.reference_stats = self._compute_stats(reference_data)
    
    def _compute_stats(self, df: pd.DataFrame) -> dict:
        """Compute summary statistics for each column"""
        stats = {}
        for col in df.columns:
            if df[col].dtype in ['int64', 'float64']:
                stats[col] = {
                    'mean': df[col].mean(),
                    'std': df[col].std(),
                    'min': df[col].min(),
                    'max': df[col].max(),
                    'percentiles': df[col].quantile([0.25, 0.5, 0.75]).tolist(),
                    'histogram': np.histogram(df[col].dropna(), bins=20),
                }
        return stats
    
    def detect_drift_ks_test(self, current_data: pd.DataFrame, 
                              feature: str, threshold: float = 0.05) -> dict:
        """
        Use Kolmogorov-Smirnov test to detect distribution shift.
        Returns test statistic and whether drift is detected.
        """
        ref = self.reference[feature].dropna()
        curr = current_data[feature].dropna()
        
        statistic, p_value = stats.ks_2samp(ref, curr)
        
        return {
            'feature': feature,
            'test': 'kolmogorov_smirnov',
            'statistic': statistic,
            'p_value': p_value,
            'drift_detected': p_value < threshold,
            'threshold': threshold,
        }
    
    def detect_drift_psi(self, current_data: pd.DataFrame, 
                         feature: str, threshold: float = 0.2) -> dict:
        """
        Population Stability Index (PSI) for drift detection.
        PSI < 0.1: No significant drift
        PSI 0.1-0.2: Moderate drift
        PSI > 0.2: Significant drift
        """
        ref = self.reference[feature].dropna()
        curr = current_data[feature].dropna()
        
        # Create bins from reference distribution
        bins = np.histogram_bin_edges(ref, bins=10)
        
        ref_hist, _ = np.histogram(ref, bins=bins)
        curr_hist, _ = np.histogram(curr, bins=bins)
        
        # Normalize to proportions
        ref_pct = ref_hist / len(ref) + 0.0001  # Avoid division by zero
        curr_pct = curr_hist / len(curr) + 0.0001
        
        # Calculate PSI
        psi = np.sum((curr_pct - ref_pct) * np.log(curr_pct / ref_pct))
        
        return {
            'feature': feature,
            'test': 'psi',
            'psi_value': psi,
            'drift_detected': psi > threshold,
            'threshold': threshold,
            'severity': 'high' if psi > 0.2 else 'moderate' if psi > 0.1 else 'low',
        }
    
    def detect_drift_mean_std(self, current_data: pd.DataFrame, 
                               feature: str, std_threshold: float = 3.0) -> dict:
        """
        Simple drift detection based on mean shift in standard deviations.
        Alerts if current mean is more than N standard deviations from reference.
        """
        ref_mean = self.reference_stats[feature]['mean']
        ref_std = self.reference_stats[feature]['std']
        
        curr_mean = current_data[feature].mean()
        
        z_score = abs(curr_mean - ref_mean) / (ref_std + 0.0001)
        
        return {
            'feature': feature,
            'test': 'mean_shift',
            'reference_mean': ref_mean,
            'current_mean': curr_mean,
            'z_score': z_score,
            'drift_detected': z_score > std_threshold,
            'threshold': std_threshold,
        }
 
# Usage in production monitoring
def monitor_feature_drift(store, feature_view: str, reference_df: pd.DataFrame):
    """Run drift detection as part of regular monitoring"""
    
    detector = FeatureDriftDetector(reference_df)
    current_df = get_recent_feature_data(store, feature_view)
    
    drift_results = []
    for feature in current_df.columns:
        if feature in reference_df.columns:
            result = detector.detect_drift_psi(current_df, feature)
            drift_results.append(result)
            
            if result['drift_detected']:
                alert_feature_drift(feature_view, result)
    
    return drift_results

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

from dataclasses import dataclass
from typing import List, Callable
import pandas as pd
from enum import Enum
 
class ValidationResult(Enum):
    PASS = "pass"
    WARN = "warn"
    FAIL = "fail"
 
@dataclass
class PipelineValidation:
    name: str
    validate_fn: Callable[[pd.DataFrame], Tuple[ValidationResult, str]]
    on_fail: str  # 'block', 'warn', 'quarantine'
 
class FeaturePipelineValidator:
    """
    Validate feature data before writing to feature store.
    Blocks bad data from contaminating the store.
    """
    
    def __init__(self):
        self.validations: List[PipelineValidation] = []
    
    def add_validation(self, validation: PipelineValidation):
        self.validations.append(validation)
    
    def validate(self, df: pd.DataFrame) -> Tuple[bool, List[dict]]:
        """
        Run all validations. Returns (should_proceed, results).
        """
        results = []
        should_proceed = True
        
        for validation in self.validations:
            try:
                result, message = validation.validate_fn(df)
                results.append({
                    'validation': validation.name,
                    'result': result.value,
                    'message': message,
                    'on_fail': validation.on_fail,
                })
                
                if result == ValidationResult.FAIL:
                    if validation.on_fail == 'block':
                        should_proceed = False
                    elif validation.on_fail == 'quarantine':
                        # Write to quarantine for investigation
                        quarantine_data(df, validation.name)
                        should_proceed = False
                        
            except Exception as e:
                results.append({
                    'validation': validation.name,
                    'result': 'error',
                    'message': str(e),
                    'on_fail': validation.on_fail,
                })
                should_proceed = False
        
        return should_proceed, results
 
# Define pipeline validations
validator = FeaturePipelineValidator()
 
# Row count validation - detect empty or truncated batches
validator.add_validation(PipelineValidation(
    name="minimum_row_count",
    validate_fn=lambda df: (
        (ValidationResult.PASS, f"Row count: {len(df)}") 
        if len(df) >= 1000 
        else (ValidationResult.FAIL, f"Only {len(df)} rows - expected >= 1000")
    ),
    on_fail="block",
))
 
# Schema validation - ensure expected columns exist
expected_columns = ['user_id', 'event_timestamp', 'purchase_count', 'total_amount']
validator.add_validation(PipelineValidation(
    name="schema_check",
    validate_fn=lambda df: (
        (ValidationResult.PASS, "Schema valid")
        if all(col in df.columns for col in expected_columns)
        else (ValidationResult.FAIL, f"Missing columns: {set(expected_columns) - set(df.columns)}")
    ),
    on_fail="block",
))
 
# Null rate validation
validator.add_validation(PipelineValidation(
    name="null_rate_check",
    validate_fn=lambda df: (
        (ValidationResult.PASS, f"Max null rate: {df.isnull().mean().max():.2%}")
        if df.isnull().mean().max() < 0.10
        else (ValidationResult.WARN, f"High null rate: {df.isnull().mean().max():.2%}")
    ),
    on_fail="warn",
))
 
# Value range validation
validator.add_validation(PipelineValidation(
    name="value_sanity_check",
    validate_fn=lambda df: (
        (ValidationResult.PASS, "Values in expected range")
        if (df['total_amount'] >= 0).all() and (df['purchase_count'] >= 0).all()
        else (ValidationResult.FAIL, "Negative values detected")
    ),
    on_fail="quarantine",
))
 
# Integration into feature pipeline
def run_feature_pipeline(source_df: pd.DataFrame):
    """Run feature transformation with validation gate"""
    
    # Transform features
    features_df = transform_features(source_df)
    
    # Validate before writing
    should_proceed, validation_results = validator.validate(features_df)
    
    if not should_proceed:
        print("Pipeline blocked due to validation failures:")
        for r in validation_results:
            if r['result'] in ['fail', 'error']:
                print(f"  - {r['validation']}: {r['message']}")
        raise Exception("Pipeline validation failed")
    
    # Write to feature store
    store.write_to_online_store(feature_view_name, features_df)
    
    return validation_results

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

from prometheus_client import Counter, Gauge, Histogram, Summary
from datetime import datetime
import logging
 
# Define comprehensive metrics
class FeatureStoreMetrics:
    def __init__(self):
        # Freshness metrics
        self.last_materialization = Gauge(
            'feature_last_materialization_timestamp',
            'Timestamp of last successful materialization',
            ['feature_view']
        )
        
        self.materialization_lag = Gauge(
            'feature_materialization_lag_seconds',
            'Seconds since last materialization',
            ['feature_view']
        )
        
        # Quality metrics
        self.null_rate = Gauge(
            'feature_null_rate',
            'Percentage of null values',
            ['feature_view', 'feature_name']
        )
        
        self.drift_score = Gauge(
            'feature_drift_psi',
            'Population Stability Index',
            ['feature_view', 'feature_name']
        )
        
        # Performance metrics
        self.query_latency = Histogram(
            'feature_online_query_latency_seconds',
            'Online store query latency',
            ['feature_view'],
            buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
        )
        
        # Error metrics
        self.pipeline_errors = Counter(
            'feature_pipeline_errors_total',
            'Total pipeline errors',
            ['feature_view', 'error_type']
        )
        
        self.validation_failures = Counter(
            'feature_validation_failures_total',
            'Total validation failures',
            ['feature_view', 'validation_name']
        )
 
metrics = FeatureStoreMetrics()
 
# Alert configuration
class AlertConfig:
    FRESHNESS_THRESHOLD_HOURS = 2
    NULL_RATE_THRESHOLD = 0.10
    DRIFT_PSI_THRESHOLD = 0.20
    LATENCY_P99_THRESHOLD_MS = 50
    MISMATCH_RATE_THRESHOLD = 0.01
 
def run_monitoring_job(store, feature_view: str, reference_df: pd.DataFrame):
    """Comprehensive monitoring job for a feature view"""
    
    fv = store.get_feature_view(feature_view)
    current_df = get_recent_feature_data(store, feature_view)
    alerts = []
    
    # Freshness check
    last_mat = get_last_materialization_time(store, feature_view)
    lag_hours = (datetime.now() - last_mat).total_seconds() / 3600
    metrics.materialization_lag.labels(feature_view=feature_view).set(lag_hours * 3600)
    
    if lag_hours > AlertConfig.FRESHNESS_THRESHOLD_HOURS:
        alerts.append({
            'type': 'freshness',
            'severity': 'high',
            'message': f'{feature_view} materialization lag: {lag_hours:.1f} hours',
        })
    
    # Quality checks per feature
    for feature in [f.name for f in fv.schema]:
        # Null rate
        null_rate = current_df[feature].isna().mean()
        metrics.null_rate.labels(feature_view=feature_view, feature_name=feature).set(null_rate)
        
        if null_rate > AlertConfig.NULL_RATE_THRESHOLD:
            alerts.append({
                'type': 'quality',
                'severity': 'medium',
                'message': f'{feature_view}.{feature} null rate: {null_rate:.1%}',
            })
        
        # Drift detection
        if feature in reference_df.columns:
            drift = calculate_psi(reference_df[feature], current_df[feature])
            metrics.drift_score.labels(feature_view=feature_view, feature_name=feature).set(drift)
            
            if drift > AlertConfig.DRIFT_PSI_THRESHOLD:
                alerts.append({
                    'type': 'drift',
                    'severity': 'high',
                    'message': f'{feature_view}.{feature} PSI: {drift:.3f}',
                })
    
    # Send alerts
    for alert in alerts:
        send_alert(alert)
    
    return alerts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

from datetime import datetime, timedelta
import logging
 
logger = logging.getLogger(__name__)
 
class FeatureStoreRecovery:
    def __init__(self, store):
        self.store = store
    
    def rollback_materialization(self, feature_view: str, 
                                   to_timestamp: datetime):
        """
        Roll back online store to a previous point in time.
        Useful when bad data was ingested.
        """
        logger.warning(f"Rolling back {feature_view} to {to_timestamp}")
        
        # Clear current online store data
        self._clear_online_store(feature_view)
        
        # Re-materialize from offline store up to the safe timestamp
        self.store.materialize(
            start_date=datetime(2020, 1, 1),  # Or earliest data date
            end_date=to_timestamp,
            feature_views=[feature_view],
        )
        
        logger.info(f"Rollback complete for {feature_view}")
    
    def backfill_time_range(self, feature_view: str,
                             start: datetime, end: datetime):
        """
        Re-compute features for a specific time range.
        Used after fixing pipeline bugs or source data issues.
        """
        logger.info(f"Backfilling {feature_view} from {start} to {end}")
        
        # First, remove any existing data in this range
        self._delete_range(feature_view, start, end)
        
        # Re-materialize
        self.store.materialize(
            start_date=start,
            end_date=end,
            feature_views=[feature_view],
        )
        
        # Validate the backfill
        validation = validate_cross_store_consistency(self.store, feature_view)
        if not validation['consistent']:
            logger.error(f"Backfill validation failed: {validation}")
            raise Exception("Backfill did not resolve consistency issues")
        
        logger.info(f"Backfill complete and validated for {feature_view}")
    
    def emergency_fallback(self, feature_view: str):
        """
        Enable emergency fallback mode when feature data is unreliable.
        Serves default values or stale data with warnings.
        """
        logger.critical(f"Enabling emergency fallback for {feature_view}")
        
        # Mark feature view as degraded in registry
        self._set_degraded_mode(feature_view, True)
        
        # Notify consumers
        notify_consumers(
            feature_view,
            message=f"DEGRADED: {feature_view} serving fallback values. "
                    f"Do not rely on these features for critical decisions."
        )
        
        # Return fallback configuration
        return {
            'mode': 'fallback',
            'feature_view': feature_view,
            'behavior': 'serve_stale_with_warning',
            'timestamp': datetime.now().isoformat(),
        }
    
    def restore_from_backup(self, feature_view: str, backup_id: str):
        """
        Restore feature data from a point-in-time backup.
        Last resort when re-computation isn't possible.
        """
        logger.warning(f"Restoring {feature_view} from backup {backup_id}")
        
        # Load backup
        backup_data = load_backup(backup_id)
        
        # Validate backup integrity
        if not validate_backup_integrity(backup_data):
            raise Exception(f"Backup {backup_id} failed integrity check")
        
        # Restore to online store
        self._restore_online_store(feature_view, backup_data)
        
        logger.info(f"Restore complete for {feature_view} from {backup_id}")