Feast: 开源特征存储亚秒级特征服务 — 2026 完整部署指南

# feature_store.yaml — Feast 项目配置
project: fraud_detection
provider: local
registry: 
  path: s3://my-bucket/registry.db  # 生产环境使用 SQL 注册中心
online_store:
  type: redis
  connection_string: "redis://localhost:6379"
offline_store:
  type: bigquery
  project: my-gcp-project
  dataset: feast_offline
entity_key_serialization_version: 2

# 检索历史特征用于训练
from feast import FeatureStore

store = FeatureStore(repo_path=".")

historical_df = store.get_historical_features(
    entity_df=entity_df,  # 包含实体 ID 和时间戳的 DataFrame
    features=[
        "user_features:avg_order_amount_30d",
        "user_features:total_transactions_90d",
        "user_features:days_since_last_order",
    ],
).to_df()

# 检索在线特征用于实时推理
features = store.get_online_features(
    features=[
        "user_features:avg_order_amount_30d",
        "user_features:total_transactions_90d",
    ],
    entity_rows=[{"user_id": "user_12345"}],
).to_dict()

# 返回: {'avg_order_amount_30d': [245.50], 'total_transactions_90d': [12]}

# 启动特征服务器
feast serve --port 6566

# 特征检索的 REST API 端点
curl -X POST "http://localhost:6566/get-online-features" \
  -H "Content-Type: application/json" \
  -d '{
    "features": ["user_features:avg_order_amount_30d"],
    "entities": {"user_id": ["user_12345"]}
  }'

# 核心 Feast (最小安装)
pip install feast

# 带 BigQuery 离线存储
pip install "feast[bigquery]"

# 带 Redis 在线存储
pip install "feast[redis]"

# 带 Snowflake
pip install "feast[snowflake]"

# 带 PostgreSQL
pip install "feast[postgres]"

# 全量安装所有常用后端
pip install "feast[gcp,redis,postgres,snowflake]"

feast version
# Feast SDK Version: 0.63.0

# 创建并进入项目目录
mkdir fraud_detection_feature_store
cd fraud_detection_feature_store

# 初始化 Feast (创建 feature_store.yaml 和 example/)
feast init

# 项目结构：
# .
# ├── feature_store.yaml    # 主配置
# ├── example/
# │   ├── repo/
# │   │   ├── example.py    # 特征定义
# │   │   └── test_workflow.py

# features/entities.py
from feast import Entity, ValueType

# 为我们的欺诈模型定义主实体
user = Entity(
    name="user_id",
    value_type=ValueType.STRING,
    description="每个用户的唯一标识符",
    join_key="user_id",
)

# features/data_sources.py
from feast import BigQuerySource

# 离线存储的历史数据源
transaction_stats_source = BigQuerySource(
    name="transaction_stats",
    query="""
        SELECT 
            user_id,
            event_timestamp,
            avg_order_amount_30d,
            total_transactions_90d,
            days_since_last_order,
            unique_merchants_30d,
            avg_transaction_amount_7d,
            failed_transaction_rate_30d,
            created
        FROM `my-gcp-project.featds.transaction_aggregates`
    """,
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

# features/feature_views.py
from feast import FeatureView, Field
from feast.types import Float32, Int64, Float64
from datetime import timedelta
from features.entities import user
from features.data_sources import transaction_stats_source

# 滑动窗口聚合的特征视图
user_transaction_features = FeatureView(
    name="user_transaction_features",
    entities=[user],
    ttl=timedelta(days=90),  # 特征有效期 90 天
    schema=[
        Field(name="avg_order_amount_30d", dtype=Float64),
        Field(name="total_transactions_90d", dtype=Int64),
        Field(name="days_since_last_order", dtype=Int64),
        Field(name="unique_merchants_30d", dtype=Int64),
        Field(name="avg_transaction_amount_7d", dtype=Float64),
        Field(name="failed_transaction_rate_30d", dtype=Float32),
    ],
    online=True,   # 从在线存储提供服务 (Redis)
    source=transaction_stats_source,
    tags={"team": "fraud", "domain": "transactions"},
    owner="ml-team@company.com",
)

# features/feature_services.py
from feast import FeatureService
from features.feature_views import user_transaction_features

# 特征服务 — 你的模型使用的接口
fraud_detection_v1 = FeatureService(
    name="fraud_detection_v1",
    features=[user_transaction_features],
    tags={"version": "1.0", "model": "fraud_xgboost"},
    owner="ml-team@company.com",
)

# 将特征定义应用到注册中心
feast apply

# 将特征从离线物化到在线存储
# (用最新特征值填充 Redis)
feast materialize-incremental $(date -u +"%Y-%m-%dT%H:%M:%S")

# 或物化特定时间范围
feast materialize 2026-01-01T00:00:00 2026-05-19T00:00:00

# feature_store.yaml — 生产配置
project: fraud_detection
provider: gcp
registry:
  registry_store_type: sql
  path: "postgresql://user:pass@pg-host:5432/feast_registry"
  cache_ttl_seconds: 60

online_store:
  type: redis
  connection_string: "redis://:password@redis-cluster.internal:6379"
  key_ttl_seconds: 604800  # 特征键的 7 天 TTL
  
offline_store:
  type: bigquery
  project: my-gcp-project
  dataset: feast_offline
  location: US

entity_key_serialization_version: 2

flags:
  alpha_features: true
  on_demand_transforms: true

# Redis Cluster 配置
online_store:
  type: redis
  redis_type: redis_cluster
  connection_string: "redis://redis-node-1:6379,redis-node-2:6379,redis-node-3:6379"
  key_ttl_seconds: 604800

# 在 Ubuntu 22.04 上部署 Redis (DigitalOcean Droplet)
sudo apt update
sudo apt install redis-server

# 生产环境配置
sudo tee -a /etc/redis/redis.conf <<EOF
maxmemory 2gb
maxmemory-policy allkeys-lru
bind 0.0.0.0
protected-mode yes
requirepass your_secure_password
EOF

sudo systemctl restart redis

# 验证
redis-cli ping
# PONG

# training_pipeline.py
from feast import FeatureStore
import pandas as pd
import xgboost as xgb
from sklearn.model_selection import train_test_split

# 初始化特征存储
store = FeatureStore(repo_path=".")

# 加载带标签的实体 DataFrame (user_id + 目标时间戳 + 标签)
entity_df = pd.read_parquet("s3://training-data/labeled_users.parquet")

# 检索时间点正确的特征
feature_service = store.get_feature_service("fraud_detection_v1")
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=feature_service,
).to_df()

# 划分和训练
X = training_df.drop(columns=["user_id", "event_timestamp", "is_fraud"])
y = training_df["is_fraud"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

model = xgb.XGBClassifier(
    max_depth=6,
    learning_rate=0.1,
    n_estimators=200,
    subsample=0.8,
)
model.fit(X_train, y_train)

# 评估
from sklearn.metrics import roc_auc_score
print(f"AUC-ROC: {roc_auc_score(y_test, model.predict_proba(X_test)[:, 1]):.4f}")

# inference_service.py
from feast import FeatureStore
from fastapi import FastAPI
import xgboost as xgb
import joblib

app = FastAPI()
store = FeatureStore(repo_path=".")
model = joblib.load("models/fraud_xgboost.pkl")

@app.post("/predict")
async def predict(user_id: str, transaction_amount: float):
    # 从 Redis 检索在线特征 (< 5ms)
    features = store.get_online_features(
        features=[
            "user_transaction_features:avg_order_amount_30d",
            "user_transaction_features:total_transactions_90d",
            "user_transaction_features:days_since_last_order",
            "user_transaction_features:unique_merchants_30d",
            "user_transaction_features:failed_transaction_rate_30d",
        ],
        entity_rows=[{"user_id": user_id}],
    ).to_dict()
    
    # 构建特征向量
    feature_vector = [
        transaction_amount,
        features["avg_order_amount_30d"][0],
        features["total_transactions_90d"][0],
        features["days_since_last_order"][0],
        features["unique_merchants_30d"][0],
        features["failed_transaction_rate_30d"][0],
    ]
    
    # 预测
    fraud_probability = model.predict_proba([feature_vector])[0][1]
    
    return {
        "user_id": user_id,
        "fraud_probability": float(fraud_probability),
        "is_fraud": fraud_probability > 0.7,
        "features_retrieved": {k: v[0] for k, v in features.items()},
    }

# dags/feast_materialize.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "ml-team",
    "depends_on_past": False,
    "email_on_failure": True,
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    "feast_materialize",
    default_args=default_args,
    schedule_interval="@hourly",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    
    materialize = BashOperator(
        task_id="materialize_features",
        bash_command="""
            cd /opt/feast/fraud_detection_feature_store && \
            feast materialize-incremental {{ ds }}T{{ ts_nodash_with_tz }}
        """,
    )
    
    validate = BashOperator(
        task_id="validate_online_store",
        bash_command="""
            cd /opt/feast/fraud_detection_feature_store && \
            python scripts/validate_online_features.py
        """,
    )
    
    materialize >> validate

from feast import on_demand_feature_view
from feast.types import Float64

# 定义按需转换
@on_demand_feature_view(
    sources=[user_transaction_features],
    schema=[
        Field(name="transaction_amount_ratio", dtype=Float64),
    ],
    mode="python",
)
def transaction_transforms(inputs):
    import pandas as pd
    df = pd.DataFrame()
    df["transaction_amount_ratio"] = (
        inputs["transaction_amount"] / inputs["avg_order_amount_30d"]
    ).fillna(0)
    return df

from feast.dqm.profilers.ge_profiler import GeProfiler

# 将数据质量期望附加到特征视图
user_transaction_features_with_validation = FeatureView(
    name="user_transaction_features",
    entities=[user],
    schema=[
        Field(name="avg_order_amount_30d", dtype=Float64),
        Field(name="total_transactions_90d", dtype=Int64),
    ],
    source=transaction_stats_source,
    profiler=GeProfiler(
        expectations=[
            {
                "expectation_type": "expect_column_mean_to_be_between",
                "kwargs": {
                    "column": "avg_order_amount_30d",
                    "min_value": 10.0,
                    "max_value": 10000.0,
                },
            },
            {
                "expectation_type": "expect_column_values_to_be_between",
                "kwargs": {
                    "column": "total_transactions_90d",
                    "min_value": 0,
                    "max_value": 10000,
                },
            },
        ]
    ),
)

# feature_store_team_a.yaml
project: team_a_fraud
registry:
  path: s3://shared-bucket/registry_team_a.db
online_store:
  type: redis
  connection_string: "redis://shared-redis:6379/0"
offline_store:
  type: bigquery
  project: my-gcp-project
  dataset: team_a_features

# RBAC 配置 (Feast 0.60+)
auth:
  type: oidc
  oidc_server_url: "https://auth.company.com"
  client_id: "feast-app"
  client_secret: "${OIDC_CLIENT_SECRET}"

dvc gc --workspace

pip install feast[redis,bigquery]
feast init
# 定义你的实体、特征视图和特征服务
feast apply
feast materialize-incremental $(date -u +"%Y-%m-%dT%H:%M:%S")