特征存储（Feature Store）架构：从一致性到在线检索的完整实践

# feature_definition.py
from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, Int64
from datetime import timedelta

customer = Entity(name="customer", join_keys=["customer_id"], value_type=Int64)

customer_stats = FeatureView(
    name="customer_stats",
    entities=[customer],
    ttl=timedelta(days=30),           # 在线存储过期时间
    schema=[
        Field(name="lifetime_value", dtype=Float32),
        Field(name="order_count_30d", dtype=Int64),
    ],
    source=FileSource(
        path="s3://features/customer_stats.parquet",
        timestamp_field="event_ts",
    ),
    online=True,                       # 同时写入在线存储
)

# training_data.py
from feast import FeatureStore
from datetime import datetime

store = FeatureStore(repo_path="feature_repo")

# 只有 label 是"未来"，所有特征都取自 label 之前的最新值
entity_df = store.get_historical_features(
    entity_df="SELECT customer_id, order_ts AS event_timestamp, label FROM training_entities",
    features=[
        "customer_stats:lifetime_value",
        "customer_stats:order_count_30d",
        "user_profile:tier",
    ],
).to_df()

# feature_repo/features.py
from feast import Entity, FeatureView, Field
from feast.infra.offline_stores.file_source import FileSource

order = Entity(name="order", join_keys=["order_id"], value_type=str)
user = Entity(name="user", join_keys=["user_id"], value_type=str)

order_features = FeatureView(
    name="order_features",
    entities=[order],
    ttl=timedelta(days=7),
    schema=[
        Field(name="amount", dtype=Float32),
        Field(name="status", dtype=str),
    ],
    source=FileSource(path="s3://features/orders.parquet", timestamp_field="ts"),
)

# 特征组：同一主题的特征放在一起便于管理与权限控制
user_features = FeatureView(
    name="user_features",
    entities=[user],
    ttl=timedelta(days=90),
    schema=[
        Field(name="tier", dtype=str),
        Field(name="churn_score", dtype=Float32),
    ],
    source=FileSource(path="s3://features/users.parquet", timestamp_field="ts"),
)

# 应用特征定义到特征存储（注册到 Registry）
feast apply

# 物化历史特征到在线存储（回填）
feast materialize-incremental 2026-09-01T00:00:00

# 查看已注册特征与版本
feast feature-views list
feast registry-dump | jq '.featureViews[].name'

# feature_store.yaml
project: rec_features
registry: s3://features/registry.db
provider: aws

online_store:
  type: redis
  connection_string: redis-cluster.xxxx.ap-southeast-1.amazonaws.com:6379
  key_ttl_seconds: 2592000   # 30 天

offline_store:
  type: file
  path: s3://features/offline

# online_serving.py
from feast import FeatureStore

store = FeatureStore(repo_path="feature_repo")

# 批量在线取特征：推荐候选集评分前统一拉取
features = store.get_online_features(
    features=[
        "user_features:tier",
        "user_features:churn_score",
        "order_features:amount_rolling_7d",
    ],
    entity_rows=[
        {"user_id": 1001, "order_id": 99881},
        {"user_id": 1002, "order_id": 99882},
    ],
).to_dict()

print(features)

# vector_features.py
from pymilvus import Collection, connections

connections.connect(host="milvus", port="19530")
col = Collection("user_embedding_v3")

result = col.search(
    data=[query_embedding],
    anns_field="embedding",
    param={"metric_type": "IP", "params": {"nprobe": 16}},
    limit=100,
    output_fields=["user_id"],
)
candidate_ids = [hit.entity.get("user_id") for hit in result[0]]

-- flink_feature.sql
CREATE TABLE cart_events (
  user_id BIGINT,
  item_id BIGINT,
  event_time TIMESTAMP(3),
  WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH ('connector' = 'kafka', 'topic' = 'cart_events', 'format' = 'json');

CREATE TABLE online_features (
  user_id BIGINT PRIMARY KEY NOT ENFORCED,
  cart_count_10m BIGINT,
  compute_time TIMESTAMP(3)
) WITH ('connector' = 'jdbc', 'url' = 'jdbc:redis://redis:6379');

INSERT INTO online_features
SELECT
  user_id,
  COUNT(*) AS cart_count_10m,
  CURRENT_TIMESTAMP AS compute_time
FROM TABLE(TUMBLE(TABLE cart_events, DESCRIPTOR(event_time), INTERVAL '1' MINUTE))
GROUP BY user_id, TUMBLE(event_time, INTERVAL '1' MINUTE);

# reconcile_features.py
def reconcile(user_id: int, batch_cnt: int, stream_cnt: int, tol: float = 0.05):
    diff = abs(batch_cnt - stream_cnt) / max(batch_cnt, 1)
    if diff > tol:
        emit_alert(f"feature skew user={user_id} batch={batch_cnt} stream={stream_cnt}")
    return diff

# feast_serving.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: feast-serving
spec:
  replicas: 3
  selector:
    matchLabels: {app: feast-serving}
  template:
    metadata:
      labels: {app: feast-serving}
    spec:
      containers:
        - name: feast-serving
          image: feastdev/feature-server:0.40
          args: ["feast", "serve", "-h", "0.0.0.0", "-p", "6566"]
          env:
            - name: FEATURE_STORE_YAML
              value: /mnt/feature_store.yaml
          ports:
            - containerPort: 6566

# minimal_fs.py
class MinimalFeatureStore:
    def __init__(self, redis, offline):
        self.redis, self.offline = redis, offline

    def register(self, name: str, sql: str):
        self.offline.save_definition(name, sql)

    def materialize(self, name: str, ts_from: str):
        df = self.offline.run(self.offline.get_definition(name), ts_from)
        for row in df.to_dict("records"):
            self.redis.set(f"feat:{name}:{row['user_id']}", encode(row))

    def get(self, name: str, keys: list):
        pipe = self.redis.pipeline()
        for k in keys:
            pipe.get(f"feat:{name}:{k}")
        return [decode(v) for v in pipe.execute()]

{
  "feature": "user_features:churn_score",
  "version": "v3",
  "owner": "growth-ml",
  "lineage": {
    "upstream": [
      {"type": "table", "name": "dws.user_health"},
      {"type": "stream", "name": "kafka:user_actions"},
      {"type": "job", "name": "spark:churn_score_daily"}
    ],
    "downstream": [
      {"type": "model", "name": "churn_prediction_v2"},
      {"type": "serving", "name": "rec-service"}
    ]
  },
  "owner_of_record": "growth-ml",
  "sla": {"freshness_minutes": 30}
}

# psi_monitor.py
import numpy as np

def compute_psi(expected: np.ndarray, actual: np.ndarray, buckets: int = 10) -> float:
    e_hist, _ = np.histogram(expected, bins=buckets)
    a_hist, _ = np.histogram(actual, bins=buckets, range=(expected.min(), expected.max()))
    e_ratio = np.clip(e_hist / e_hist.sum(), 1e-6, 1)
    a_ratio = np.clip(a_hist / a_hist.sum(), 1e-6, 1)
    return float(np.sum((a_ratio - e_ratio) * np.log(a_ratio / e_ratio)))

print("PSI:", round(compute_psi(train_churn_score, online_churn_score), 3))

# ranking_features.py
from feast import FeatureStore

store = FeatureStore(repo_path="feature_repo")

def get_ranking_features(user_id: int, candidate_items: list) -> dict:
    # 用户侧特征 + 物品侧特征 + 交叉特征一次取回
    online = store.get_online_features(
        features=[
            "user_features:recent_cat_weights",
            "user_features:avg_ctr_7d",
            "item_features:item_embedding",
            "item_features:item_cat",
        ],
        entity_rows=[{"user_id": user_id, "item_id": i} for i in candidate_items],
    ).to_dict()
    return {k: online[k] for k in ["item_embedding", "avg_ctr_7d", "item_cat"]}

# feature_backtest.py
def backtest_feature(store, feature_name: str, entity_df):
    # 使用历史时间点重放特征，评估特征有效性
    hist = store.get_historical_features(
        entity_df=entity_df,
        features=[f"{feature_name}"],
    ).to_df()
    return evaluate_importance(hist)