AI Platform Feature Store 同步设计

from feast import FeatureStore
from datetime import datetime, timedelta
import pandas as pd

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

# Entity DataFrame：预测对象 + 预测时刻
entity_df = pd.DataFrame({
    "user_id": ["u_001", "u_002", "u_003"],
    "event_timestamp": [
        datetime(2026, 3, 3, 10, 0, 0),  # 每个用户的预测时刻各不相同
        datetime(2026, 3, 3, 14, 30, 0),
        datetime(2026, 3, 4, 9, 0, 0),
    ],
})

# Feast 会自动匹配每一行 event_timestamp 之前发生的最新特征
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_purchase_stats:total_purchases_7d",
        "user_purchase_stats:avg_order_value_30d",
        "user_session_features:session_count_24h",
        "user_session_features:last_active_minutes_ago",
    ],
).to_df()

print(training_df.head())
# user_id | event_timestamp | total_purchases_7d | avg_order_value_30d | ...

import pandas as pd
from typing import List

def validate_point_in_time_correctness(
    training_df: pd.DataFrame,
    entity_timestamp_col: str,
    feature_timestamp_cols: List[str],
    tolerance_seconds: int = 0,
) -> dict:
    """
    检测训练数据中，特征时间戳晚于(未来于)实体时间戳的行。
    可通过 tolerance_seconds 允许一定的时钟漂移。
    """
    violations = {}
    for feat_ts_col in feature_timestamp_cols:
        if feat_ts_col not in training_df.columns:
            continue
        mask = (
            training_df[feat_ts_col]
            > training_df[entity_timestamp_col] + pd.Timedelta(seconds=tolerance_seconds)
        )
        violation_count = mask.sum()
        if violation_count > 0:
            violations[feat_ts_col] = {
                "count": int(violation_count),
                "ratio": round(violation_count / len(training_df), 4),
                "sample_entity_ts": str(training_df.loc[mask, entity_timestamp_col].iloc[0]),
                "sample_feat_ts": str(training_df.loc[mask, feat_ts_col].iloc[0]),
            }
    return violations  # 为空表示正常

# 在 CI 中调用
result = validate_point_in_time_correctness(
    training_df,
    entity_timestamp_col="event_timestamp",
    feature_timestamp_cols=["feature_timestamp_purchase", "feature_timestamp_session"],
    tolerance_seconds=5,  # 允许 5 秒时钟漂移
)
if result:
    raise AssertionError(f"Point-in-time leakage detected: {result}")

# feature_store.yaml(Feast 0.40+ 配置)
project: recommendation_platform
provider: gcp
registry:
  registry_type: sql
  path: postgresql://feast:feast@pg-host:5432/feast_registry
  cache_ttl_seconds: 60

offline_store:
  type: bigquery
  dataset: features_offline

online_store:
  type: redis
  connection_string: redis://redis-cluster:6379
  key_ttl_seconds: 86400 # 24 小时 TTL

# 同步配置
stream_ingestion:
  enabled: true
  kafka:
    bootstrap_servers: kafka-broker-1:9092,kafka-broker-2:9092
    topic_prefix: feast-features
    consumer_group: feast-materializer
    security_protocol: SASL_SSL
    sasl_mechanism: SCRAM-SHA-512

"""
通过 Kafka Consumer 接收 CDC 事件并写入 Feast Online Store 的 worker。
处理由 Debezium 的 PostgreSQL CDC 连接器发布的事件。
"""
import json
from datetime import datetime
from confluent_kafka import Consumer, KafkaError
from feast import FeatureStore
import pandas as pd

KAFKA_CONFIG = {
    "bootstrap.servers": "kafka-broker-1:9092",
    "group.id": "feast-cdc-materializer",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
    "max.poll.interval.ms": 300000,
}

store = FeatureStore(repo_path="./feature_repo")
consumer = Consumer(KAFKA_CONFIG)
consumer.subscribe(["dbserver1.public.user_purchases"])

BATCH_SIZE = 500
FLUSH_INTERVAL_SEC = 10
buffer = []
last_flush = datetime.now()

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None:
        pass
    elif msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        raise Exception(f"Kafka error: {msg.error()}")
    else:
        payload = json.loads(msg.value().decode("utf-8"))
        after = payload.get("after", {})
        if after:
            buffer.append({
                "user_id": after["user_id"],
                "total_purchases_7d": after["total_purchases_7d"],
                "avg_order_value_30d": after["avg_order_value_30d"],
                "event_timestamp": datetime.fromisoformat(after["updated_at"]),
            })

    elapsed = (datetime.now() - last_flush).total_seconds()
    if len(buffer) >= BATCH_SIZE or (buffer and elapsed >= FLUSH_INTERVAL_SEC):
        df = pd.DataFrame(buffer)
        # 直接写入 Feast Online Store
        store.write_to_online_store(
            feature_view_name="user_purchase_stats",
            df=df,
        )
        consumer.commit()
        print(f"Materialized {len(buffer)} rows to online store")
        buffer.clear()
        last_flush = datetime.now()

-- Online Store freshness 看板查询
-- 计算每个 entity 最后更新时间与当前时间的差值

WITH freshness AS (
    SELECT
        entity_key,
        feature_view_name,
        MAX(event_ts)                              AS latest_event_ts,
        EXTRACT(EPOCH FROM now() - MAX(event_ts))  AS lag_seconds
    FROM online_store_features
    WHERE feature_view_name = 'user_purchase_stats'
    GROUP BY entity_key, feature_view_name
)
SELECT
    feature_view_name,
    COUNT(*)                                                      AS total_entities,
    COUNT(*) FILTER (WHERE lag_seconds <= 300)                    AS fresh_entities,
    ROUND(100.0 * COUNT(*) FILTER (WHERE lag_seconds <= 300) / COUNT(*), 2) AS freshness_pct,
    ROUND(AVG(lag_seconds), 1)                                   AS avg_lag_sec,
    ROUND(PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY lag_seconds), 1) AS p95_lag_sec,
    MAX(lag_seconds)                                              AS max_lag_sec
FROM freshness
GROUP BY feature_view_name;

# prometheus-rules.yaml
groups:
  - name: feature_store_freshness
    interval: 30s
    rules:
      - alert: FeatureStoreFreshnessBreached
        expr: |
          feast_feature_view_freshness_seconds{feature_view="user_purchase_stats"} > 300
        for: 3m
        labels:
          severity: warning
          team: ml-platform
        annotations:
          summary: 'Feature View {{ $labels.feature_view }} 的 freshness SLA 违规'
          description: |
            当前 lag：{{ $value }} 秒(SLA：300 秒)。
            请检查 CDC 流水线或 Kafka consumer lag。

      - alert: FeatureStoreOnlineStoreDown
        expr: |
          up{job="feast-online-store"} == 0
        for: 1m
        labels:
          severity: critical
          team: ml-platform
        annotations:
          summary: 'Online Store(Redis) 无法连接'
          description: '正在提供服务的模型无法查询特征。需要立即确认。'

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: recommendation-model-v3
  labels:
    feature-view-version: 'v2026_03_04'
    model-version: 'v3.2.1'
  annotations:
    feast.dev/feature-views: 'user_purchase_stats:v2026_03_04,user_session_features:v2026_03_01'
spec:
  predictor:
    model:
      modelFormat:
        name: mlflow
      storageUri: gs://ml-models/recommendation/v3.2.1
    containers:
      - name: kserve-container
        env:
          - name: FEATURE_VIEW_VERSION
            value: 'v2026_03_04'
          - name: FEAST_REPO_PATH
            value: '/mnt/feast-repo'
          - name: ONLINE_STORE_TIMEOUT_MS
            value: '50'
        resources:
          requests:
            memory: '2Gi'
            cpu: '1'
          limits:
            memory: '4Gi'
            cpu: '2'

"""
验证模型所期望的特征 schema 与 Feature View 所提供的 schema
是否兼容的 CI 测试。用 pytest 运行。
"""
import pytest
from feast import FeatureStore
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class ModelFeatureContract:
    """模型所要求的特征契约"""
    model_name: str
    feature_view_version: str
    expected_features: Dict[str, str]  # feature_name -> expected_type

# 各模型的特征契约定义
CONTRACTS = [
    ModelFeatureContract(
        model_name="recommendation-model-v3",
        feature_view_version="v2026_03_04",
        expected_features={
            "total_purchases_7d": "INT64",
            "avg_order_value_30d": "DOUBLE",
            "session_count_24h": "INT64",
            "last_active_minutes_ago": "DOUBLE",
        },
    ),
]

@pytest.fixture
def feature_store():
    return FeatureStore(repo_path="./feature_repo")

@pytest.mark.parametrize("contract", CONTRACTS, ids=lambda c: c.model_name)
def test_feature_schema_compatibility(feature_store, contract):
    """验证 Feature View schema 是否与模型契约一致"""
    fv = feature_store.get_feature_view(f"user_purchase_stats_{contract.feature_view_version}")

    actual_schema = {f.name: str(f.dtype) for f in fv.features}

    for feat_name, expected_type in contract.expected_features.items():
        assert feat_name in actual_schema, (
            f"Feature '{feat_name}' not found in FeatureView. "
            f"Available: {list(actual_schema.keys())}"
        )
        assert actual_schema[feat_name] == expected_type, (
            f"Feature '{feat_name}' type mismatch: "
            f"expected {expected_type}, got {actual_schema[feat_name]}"
        )

症状：A/B 测试报告中，offline 复现数值与 online 数值相差 3% 以上
原因：服务代码里直接实现的特征变换(归一化、clipping)部分，
      与 Feast Feature View 的变换逻辑不一致
错误日志：
  AssertionError: offline_ctr=0.0823, online_ctr=0.1134, diff=0.0311

解决方法：
  1. 把所有特征变换都定义在 Feast FeatureView 或 OnDemandFeatureView 内
  2. 禁止在服务代码中直接实现变换逻辑
  3. 在 CI 中添加复现性测试(见下方代码)

def test_online_offline_parity(feature_store, sample_entities):
    """验证 Online Store 查询值与 Offline Store 复现值是否一致"""
    online_features = feature_store.get_online_features(
        features=["user_purchase_stats:total_purchases_7d"],
        entity_rows=[{"user_id": eid} for eid in sample_entities],
    ).to_dict()

    # 查询同一时刻的 Offline 值
    entity_df = pd.DataFrame({
        "user_id": sample_entities,
        "event_timestamp": [datetime.now()] * len(sample_entities),
    })
    offline_features = feature_store.get_historical_features(
        entity_df=entity_df,
        features=["user_purchase_stats:total_purchases_7d"],
    ).to_df()

    for i, eid in enumerate(sample_entities):
        online_val = online_features["total_purchases_7d"][i]
        offline_val = offline_features.loc[
            offline_features["user_id"] == eid, "total_purchases_7d"
        ].iloc[0]
        assert online_val == offline_val, (
            f"Parity violation for {eid}: online={online_val}, offline={offline_val}"
        )

症状：因 3 天的 CDC 故障而执行 backfill 后，模型 precision 从 0.82 降至 0.61
原因：backfill 时把 event_timestamp 错误地设置成了 backfill 的执行时刻，
      导致 point-in-time join 中出现未来数据 leakage
错误：没有直接报错，仅通过指标下降才被发现。

解决方法：
  1. backfill 时必须保留原始 event_timestamp
  2. backfill 完成后自动执行 leakage 验证脚本
  3. 把模型质量回归测试纳入 backfill runbook

症状：服务 pod 间歇性返回 5xx 响应。错误日志：
  redis.exceptions.TimeoutError: Timeout reading from redis-cluster:6379
  feast.errors.FeatureRetrievalError: Failed to retrieve features in 50ms

原因：替换 1 台 Redis 集群节点期间 failover 延迟

解决方法：
  1. 在 Feast online_store 配置中添加 timeout + retry
  2. 在服务代码中添加特征查询失败时的 fallback 逻辑
  3. 应用 Circuit breaker 模式(连续失败 3 次时切换到 fallback)

from tenacity import retry, stop_after_attempt, wait_exponential
from feast import FeatureStore
from typing import Dict, List, Optional

class ResilientFeatureFetcher:
    """特征查询失败时切换到 fallback 的封装类"""

    def __init__(self, store: FeatureStore, fallback_ttl_seconds: int = 300):
        self.store = store
        self.fallback_cache: Dict[str, dict] = {}
        self.consecutive_failures = 0
        self.circuit_open = False
        self.failure_threshold = 3

    @retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=0.01, max=0.05))
    def _fetch_online(self, features: List[str], entity_rows: List[dict]) -> dict:
        return self.store.get_online_features(
            features=features, entity_rows=entity_rows
        ).to_dict()

    def get_features(
        self, features: List[str], entity_rows: List[dict]
    ) -> Optional[dict]:
        if self.circuit_open:
            return self._get_fallback(entity_rows)

        try:
            result = self._fetch_online(features, entity_rows)
            self.consecutive_failures = 0
            # 成功时更新缓存
            for i, row in enumerate(entity_rows):
                key = str(row)
                self.fallback_cache[key] = {
                    f: result[f][i] for f in features
                }
            return result
        except Exception:
            self.consecutive_failures += 1
            if self.consecutive_failures >= self.failure_threshold:
                self.circuit_open = True
            return self._get_fallback(entity_rows)

    def _get_fallback(self, entity_rows: List[dict]) -> Optional[dict]:
        """从缓存中返回 fallback 值。缓存未命中时使用默认值。"""
        # 在实际服务中，应返回本地缓存或预先计算好的平均值，而不是 Redis
        return None